Added new tasks status metrics by count (#80)

* Added new task count metrics

Signed-off-by: iamabhishek-dubey <abhishekbhardwaj510@gmail.com>

* Added new task count metrics

Signed-off-by: iamabhishek-dubey <abhishekbhardwaj510@gmail.com>

Author: iamabhishek-dubey

CHANGELOG.md

@@ -10,6 +10,7 @@
 
 - Added datasource row count as metrics
 - Docker compose setup to fix
+- Added new task count metrics
 
 ### v0.8
 ##### July 11, 2020

collector/druid.go

@@ -106,6 +106,26 @@ func GetDruidDataSourcesTotalRows(pathURL string) DataSourcesTotalRows {
 	return datasources
 }
 
+// GetDruidTasksStatusCount returns count of different tasks by status
+func GetDruidTasksStatusCount(pathURL string) TaskStatusMetric {
+	kingpin.Parse()
+	druidURL := *druid + pathURL
+	responseData, err := utils.GetResponse(druidURL, pathURL)
+	if err != nil {
+		logrus.Errorf("Cannot retrieve data for druid's workers: %v", err)
+		return nil
+	}
+	logrus.Debugf("Successfully retrieved the data for druid task: %v", pathURL)
+	var taskCount TaskStatusMetric
+	err = json.Unmarshal(responseData, &taskCount)
+	if err != nil {
+		logrus.Errorf("Cannot parse JSON data: %v", err)
+		return nil
+	}
+	logrus.Debugf("Successfully collected tasks status count: %v", pathURL)
+	return taskCount
+}
+
 // getDruidWorkersData return all the workers and its state
 func getDruidWorkersData(pathURL string) []worker {
 	kingpin.Parse()
@@ -137,6 +157,10 @@ func (collector *MetricCollector) Describe(ch chan<- *prometheus.Desc) {
 	ch <- collector.DruidWorkers
 	ch <- collector.DruidTasks
 	ch <- collector.DruidSegmentReplicateSize
+	ch <- collector.DruidRunningTasks
+	ch <- collector.DruidWaitingTasks
+	ch <- collector.DruidCompletedTasks
+	ch <- collector.DruidPendingTasks
 }
 
 // Collector return the defined metrics
@@ -179,6 +203,22 @@ func Collector() *MetricCollector {
 		DruidDataSourcesTotalRows: prometheus.NewDesc("druid_datasource_total_rows",
 			"Number of rows in a datasource",
 			[]string{"datasource_name", "source"}, nil),
+		DruidRunningTasks: prometheus.NewDesc("druid_running_tasks",
+			"Druid running tasks count",
+			nil, nil,
+		),
+		DruidWaitingTasks: prometheus.NewDesc("druid_waiting_tasks",
+			"Druid waiting tasks count",
+			nil, nil,
+		),
+		DruidCompletedTasks: prometheus.NewDesc("druid_completed_tasks",
+			"Druid completed tasks count",
+			nil, nil,
+		),
+		DruidPendingTasks: prometheus.NewDesc("druid_pending_tasks",
+			"Druid pending tasks count",
+			nil, nil,
+		),
 	}
 }
 
@@ -203,6 +243,15 @@ func (collector *MetricCollector) Collect(ch chan<- prometheus.Metric) {
 		}
 	}
 
+	ch <- prometheus.MustNewConstMetric(collector.DruidRunningTasks,
+		prometheus.GaugeValue, float64(len(GetDruidTasksStatusCount(runningTask))))
+	ch <- prometheus.MustNewConstMetric(collector.DruidWaitingTasks,
+		prometheus.GaugeValue, float64(len(GetDruidTasksStatusCount(waitingTask))))
+	ch <- prometheus.MustNewConstMetric(collector.DruidCompletedTasks,
+		prometheus.GaugeValue, float64(len(GetDruidTasksStatusCount(completedTask))))
+	ch <- prometheus.MustNewConstMetric(collector.DruidPendingTasks,
+		prometheus.GaugeValue, float64(len(GetDruidTasksStatusCount(pendingTask))))
+
 	workers := getDruidWorkersData(workersURL)
 
 	for _, worker := range workers {

collector/interface.go

@@ -14,6 +14,10 @@ const (
 	workersURL     = "/druid/indexer/v1/workers"
 	supervisorURL  = "/druid/indexer/v1/supervisor?full"
 	sqlURL         = "/druid/v2/sql"
+	pendingTask    = "/druid/indexer/v1/pendingTasks"
+	runningTask    = "/druid/indexer/v1/runningTasks"
+	waitingTask    = "/druid/indexer/v1/waitingTasks"
+	completedTask  = "/druid/indexer/v1/completeTasks"
 )
 
 const totalRowsSQL = `select SEG.datasource, SUP.source,
@@ -33,6 +37,10 @@ type MetricCollector struct {
 	DruidSegmentSize          *prometheus.Desc
 	DruidSegmentReplicateSize *prometheus.Desc
 	DruidDataSourcesTotalRows *prometheus.Desc
+	DruidRunningTasks         *prometheus.Desc
+	DruidWaitingTasks         *prometheus.Desc
+	DruidCompletedTasks       *prometheus.Desc
+	DruidPendingTasks         *prometheus.Desc
 }
 
 // DataSourcesTotalRows shows total rows from each datasource
@@ -76,6 +84,12 @@ type TasksInterface []struct {
 	DataSource       string  `json:"dataSource"`
 }
 
+// TaskStatusMetric is the interface for tasks status
+type TaskStatusMetric []struct {
+	NameDataSource string `json:"dataSource"`
+	StatusCode     string `json:"statusCode"`
+}
+
 type worker struct {
 	Worker struct {
 		Host     string `json:"host"`