Add Prometheus Endpoint to the core and extensions modules of streampipes.

[luoluoyuyu]

Purpose

To support prometheus in monitoring streampipes.

Solution

Using 'spring-boot-starter-actuator'

Adds monitoring metrics to a springboot application using 'spring-boot-starter-actuator', 'spring-boot-starter-actuator' provides information about the memory usage of the springboot application, the number of threads, and information related to CPU, etc.

Adding Custom Metrics

Add custom metrics on top of the metrics provided by 'spring-boot-starter-actuator', add custom metrics about prometheus in streampipes and put this related code into the streampipes-commons module.

Specific implementation

Integrate actuator

Add actuator library and actuator configuration to StreamPipesServiceBase module to integrate actuator with core and extensions modules.

  private void runApplication(Class<?> serviceClass,
                              Integer port) {
      SpringApplication app = new SpringApplication(serviceClass);
      app.setDefaultProperties(getProperties(port));
      app.run();
  }

  private  Properties getProperties(Integer port) {
    Properties properties = new Properties();

    properties.setProperty("management.endpoints.web.exposure.include", "health,prometheus");

    properties.setProperty("management.metrics.tags.application", "${spring.application.name}");

    properties.setProperty("server.port", port.toString());

    return properties;
  } 

Custom Indicators

1、Implement a global registry

public class StreamPipesCollectorRegistry {

    private static final CollectorRegistry collectorRegistry=new CollectorRegistry(true);

    public static CollectorRegistry getCollectorRegistry(){
        return collectorRegistry;
    }
}

2, in straempipes to create a Prometheus module, in the module to define indicators such as pipline statistics class.

public class PipelinesStats {

  private int allPipelines;

  private int runningPipelines;

  private int stoppedPipelines;

  private int healthyPipelines;

  private int failedPipelines;

  private int attentionRequiredPipelines;
}

3, create a Prometheus indicator class

public class PipelinesStatsMetrics {

  public static final Gauge allPipelinesGauge = StreamPipesCollectorRegistry
                                                  .registerGauge(
                                                    "all_pipelines",
                                                    "Total number of pipelines");
//....

}

4、Data Buried Points
Buried in the execution method

public class PipelineHealthCheck implements Runnable {

  private static final PipelinesStats pipelinesStats = new PipelinesStats();


  public void checkAndRestorePipelineElements() {
    //......
    pipelinesStats.clear();
    pipelinesStats.setAllPipelines(allPipelines.size());
    pipelinesStats.setRunningPipelines(runningPipelines.size());
    pipelinesStats.setStoppedPipelines(pipelinesStats.getAllPipelines() - pipelinesStats.getRunningPipelines());

    if (runningPipelines.size() > 0) {
      //..

      runningPipelines.forEach(pipeline -> {
        //.....
        if (shouldUpdatePipeline.get()) {
          if (failedInstances.size() > 0) {
            //...
            pipelinesStats.FailedIncrease();
          } else if (recoveredInstances.size() > 0) {
            //...
            pipelinesStats.attentionRequiredIncrease();
          }
          //...
        }
      });
      int healthNum=pipelinesStats.getRunningPipelines() - pipelinesStats.getFailedPipelines() - pipelinesStats.getAttentionRequiredPipelines();
      pipelinesStats.setHealthyPipelines(healthNum);
      pipelinesStats.metrics();
    }
  }
  
}

5、Register
In the core module to register a custom CollectorRegistry, so that the CollectorRegistry can be assembled into the PrometheusMeterRegistry.

@Configuration
public class StreamPipesPrometheus {

  @Bean
  public CollectorRegistry collectorRegistry() {
    return StreamPipesCollectorRegistry.getCollectorRegistry();
  }
  
}

[dominikriemer]

HI @luoluoyuyu that's an awesome idea! I really like this a lot.
Is there anything we need to consider in terms of authentication? So that the provided endpoint is secured against unauthenticated access?

[luoluoyuyu]

hi @dominikriemer
Yes, we should think about authentication, Prometheus can get a token or a token file to pass the checksum. But the current token for streampipes has an expiration date, and Prometheus can't change the token while it's running.

[dominikriemer]

Ok great! Maybe we can use the API key feature from the profile section, which doesn't expire and uses HTTP headers X-API-USER and X-API-KEY. Not perfect since it's bound to a user but this is an option. Alternatively, we could think of removing the token expiration.

[tenthe]

Hi @luoluoyuyu,
thanks for the proposal. I also really like the idea.
I was wondering what kind of services / docker containers will we need to monitor the logs in a production system?
Cheers,
Philipp

[luoluoyuyu]

hi @tenthe
These metrics can be monitored using Grafana. Later we can integrate Prometheus and Grafana in k8s, and provide users with dashboard files to import into Grafana.
Monitoring with grafana looks like the following image

[luoluoyuyu]

hi
I submitted a pr on this proposal（#1950）