Merge pull request #441 from FlowCI/feature/observability

Feature/observability

Author: gy2006

core/debug.yaml

@@ -45,4 +45,21 @@ services:
  ports:
  - "9000:9000"
  - "9090:9090"
- command: "server /data --console-address \":9090\""
+ command: "server /data --console-address \":9090\""
+
+ prometheus:
+ image: prom/prometheus:v2.44.0
+ container_name: prometheus
+ ports:
+ - "9191:9090"
+ volumes:
+ - ./observability/prometheus/prometheus.yml:/etc/prometheus/prometheus.yml
+
+ grafana:
+ image: grafana/grafana:9.5.2
+ container_name: grafana
+ ports:
+ - "3131:3000"
+ restart: unless-stopped
+ volumes:
+ - ./observability/grafana/datasources:/etc/grafana/provisioning/datasources

core/observability/grafana/datasources/datasources.yml

@@ -0,0 +1,7 @@
+apiVersion: 1
+datasources:
+ - name: Prometheus
+ type: prometheus
+ access: proxy
+ url: http://prometheus:9090
+ isDefault: true

core/observability/prometheus/prometheus.yml

@@ -0,0 +1,8 @@
+scrape_configs:
+ - job_name: 'FlowCIMetrics'
+ metrics_path: '/actuator/prometheus'
+ scrape_interval: 5s
+ static_configs:
+ - targets: ['host.docker.internal:8080']
+ labels:
+ application: 'flow.ci'

core/pom.xml

@@ -103,6 +103,12 @@
  <artifactId>spring-boot-starter-cache</artifactId>
  </dependency>
 
+ <dependency>
+ <groupId>io.micrometer</groupId>
+ <artifactId>micrometer-registry-prometheus</artifactId>
+ <scope>runtime</scope>
+ </dependency>
+
  <dependency>
  <groupId>org.eclipse.jgit</groupId>
  <artifactId>org.eclipse.jgit</artifactId>
@@ -139,6 +145,12 @@
  <configuration>
  <mainClass>com.flowci.core.Application</mainClass>
  <layout>JAR</layout>
+ <excludes>
+ <exclude>
+ <groupId>org.projectlombok</groupId>
+ <artifactId>lombok</artifactId>
+ </exclude>
+ </excludes>
  </configuration>
  <executions>
  <execution>

core/src/main/java/com/flowci/core/agent/service/AgentServiceImpl.java

@@ -143,7 +143,7 @@ public void subscribeIdleAgentQueue() throws IOException {
  if (shouldPushBack) {
  int randomSec = ObjectsHelper.randomNumber(MinIdleAgentPushBack, MaxIdleAgentPushBack);
  ThreadHelper.sleep(randomSec * 1000L);
- idleAgentQueueManager.send(idleAgentQueue, agentId.getBytes());
+ idleAgentQueueManager.publish(idleAgentQueue, agentId.getBytes());
  }
  } catch (Exception e) {
  log.warn(e.getMessage());
@@ -307,7 +307,7 @@ public void release(Collection<String> ids) {
  update(agent, OFFLINE);
  case BUSY:
  update(agent, IDLE);
- idleAgentQueueManager.send(idleAgentQueue, agentId.getBytes());
+ idleAgentQueueManager.publish(idleAgentQueue, agentId.getBytes());
  }
  }
  } finally {
@@ -406,7 +406,7 @@ public void onConnected(OnConnectedEvent event) {
  update(target, init.getStatus());
 
  if (target.isIdle() && event.isToIdleQueue()) {
- idleAgentQueueManager.send(idleAgentQueue, target.getId().getBytes());
+ idleAgentQueueManager.publish(idleAgentQueue, target.getId().getBytes());
  }
 
  event.setAgent(target);

core/src/main/java/com/flowci/core/common/rabbit/RabbitOperations.java

@@ -25,7 +25,6 @@
 import org.springframework.core.task.TaskExecutor;
 
 import java.io.IOException;
-import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
@@ -64,19 +63,24 @@ public void declareExchangeAndBind(String exchange,
  }
 
  public void declare(String queue, boolean durable) throws IOException {
- this.channel.queueDeclare(queue, durable, false, false, null);
+ this.channel.queueDeclare(queue, durable, false, false, Map.of(
+ "x-single-active-consumer", true
+ ));
  }
 
  public void declareTemp(String queue) throws IOException {
- this.channel.queueDeclare(queue, false, false, true, null);
+ this.channel.queueDeclare(queue, false, false, true, Map.of(
+ "x-single-active-consumer", true
+ ));
  }
 
  public void declare(String queue, boolean durable, Integer maxPriority, String dlExName) throws IOException {
- Map<String, Object> props = new HashMap<>(3);
- props.put("x-max-priority", maxPriority);
- props.put("x-dead-letter-exchange", dlExName);
- props.put("x-dead-letter-routing-key", QueueConfig.JobDlRoutingKey);
- this.channel.queueDeclare(queue, durable, false, false, props);
+ this.channel.queueDeclare(queue, durable, false, false, Map.of(
+ "x-max-priority", maxPriority,
+ "x-dead-letter-exchange", dlExName,
+ "x-dead-letter-routing-key", QueueConfig.JobDlRoutingKey,
+ "x-single-active-consumer", true
+ ));
  }
 
  public boolean delete(String queue) {
@@ -108,9 +112,9 @@ public boolean sendToEx(String ex, byte[] body, Map<String, Object> headers) {
  }
 
  /**
- * Send to routing key with default exchange
+ * Publish data with routing key to default exchange
  */
- public boolean send(String routingKey, byte[] body) {
+ public boolean publish(String routingKey, byte[] body) {
  try {
  this.channel.basicPublish(StringHelper.EMPTY, routingKey, null, body);
  return true;
@@ -120,9 +124,9 @@ public boolean send(String routingKey, byte[] body) {
  }
 
  /**
- * Send to routing key with default exchange and priority
+ * Publish data with routing key and priority to default exchange
  */
- public boolean send(String routingKey, byte[] body, Integer priority, int expireInSecond) {
+ public boolean publish(String routingKey, byte[] body, Integer priority, int expireInSecond) {
  try {
  AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
  .priority(priority)

core/src/main/java/com/flowci/core/job/service/JobActionServiceImpl.java

@@ -370,7 +370,7 @@ public void accept(JobSmContext context) {
  String queue = job.getQueueName();
  byte[] payload = job.getId().getBytes();
 
- jobsQueueManager.send(queue, payload, job.getPriority(), job.getExpire());
+ jobsQueueManager.publish(queue, payload, job.getPriority(), job.getExpire());
  logInfo(job, "enqueue");
  }
 

core/src/main/java/com/flowci/core/metrics/MetricsManager.java

@@ -0,0 +1,27 @@
+package com.flowci.core.metrics;
+
+import com.flowci.core.job.event.JobFinishedEvent;
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.MeterRegistry;
+import lombok.AllArgsConstructor;
+import lombok.extern.log4j.Log4j2;
+import org.springframework.context.event.EventListener;
+import org.springframework.stereotype.Component;
+
+@Component
+@Log4j2
+@AllArgsConstructor
+public class MetricsManager {
+
+ private final MeterRegistry meterRegistry;
+
+ @EventListener(JobFinishedEvent.class)
+ public void onJobFinished(JobFinishedEvent e) {
+ Counter.builder("num_of_finished_job")
+ .description("num of finished job")
+ .register(meterRegistry)
+ .increment();
+
+ log.debug("metrics: num_of_finished_job increment");
+ }
+}

core/src/main/resources/application.properties

@@ -1,18 +1,23 @@
+spring.application.name=flow.ci
+
 logging.level.org.apache.zookeeper=ERROR
 logging.level.org.apache.curator.framework=ERROR
 logging.level.com.flowci.core=${FLOWCI_LOG_LEVEL:INFO}
 
 info.app.version=1.23.01
-info.app.name=flow.ci
+info.app.name=${spring.application.name}
 
 server.port=${FLOWCI_SERVER_PORT:8080}
 server.address=${FLOWCI_SERVER_ADDRESS:0.0.0.0}
 server.tomcat.uri-encoding=UTF-8
 
+management.metrics.tags.application=${spring.application.name}
+
 management.endpoint.health.enabled=true
 management.endpoint.health.show-details=always
 management.endpoint.shutdown.enabled=true
-management.endpoints.web.base-path=/
+management.endpoints.web.base-path=/actuator
+management.endpoints.web.exposure.include=info,prometheus,metrics,health
 
 spring.servlet.multipart.enabled=true
 spring.servlet.multipart.location=${java.io.tmpdir}