🔀 [patch] Merge pull request #30 from agoda-com/exporter

Prometheus Queue Improvement

cmd/staging/main.go

@@ -141,8 +141,8 @@ func startCtrlCmd() *cobra.Command {
 			samsahaiClient := rpc.NewRPCProtobufClient(viper.GetString(s2h.VKS2HServerURL), &http.Client{})
 			configCtrl := configctrl.New(mgr)
 			queueCtrl := queue.New(namespace, runtimeClient)
-			desiredctrl.New(teamName, mgr, queueCtrl)
 			authToken := viper.GetString(s2h.VKS2HAuthToken)
+			desiredctrl.New(teamName, mgr, queueCtrl, authToken, samsahaiClient)
 			tcBaseURL := viper.GetString(s2h.VKTeamcityURL)
 			tcUsername := viper.GetString(s2h.VKTeamcityUsername)
 			tcPassword := viper.GetString(s2h.VKTeamcityPassword)

internal/desiredcomponent/controller.go

@@ -3,7 +3,9 @@ package desiredcomponent
 import (
 	"context"
 	"fmt"
+	"net/http"
 
+	"github.com/twitchtv/twirp"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -18,6 +20,7 @@ import (
 	s2herrors "github.com/agoda-com/samsahai/internal/errors"
 	s2hlog "github.com/agoda-com/samsahai/internal/log"
 	"github.com/agoda-com/samsahai/internal/queue"
+	samsahairpc "github.com/agoda-com/samsahai/pkg/samsahai/rpc"
 )
 
 const (
@@ -30,6 +33,8 @@ type controller struct {
 	teamName  string
 	queueCtrl internal.QueueController
 	client    client.Client
+	authToken string
+	s2hClient samsahairpc.RPC
 }
 
 var _ internal.DesiredComponentController = &controller{}
@@ -38,6 +43,8 @@ func New(
 	teamName string,
 	mgr manager.Manager,
 	queueCtrl internal.QueueController,
+	authToken string,
+	s2hClient samsahairpc.RPC,
 ) internal.DesiredComponentController {
 	if queueCtrl == nil {
 		logger.Error(s2herrors.ErrInternalError, "queue ctrl cannot be nil")
@@ -48,6 +55,8 @@ func New(
 		teamName:  teamName,
 		queueCtrl: queueCtrl,
 		client:    mgr.GetClient(),
+		authToken: authToken,
+		s2hClient: s2hClient,
 	}
 
 	if err := add(mgr, c); err != nil {
@@ -108,6 +117,23 @@ func (c *controller) Reconcile(req reconcile.Request) (reconcile.Result, error)
 		return reconcile.Result{}, err
 	}
 
+	headers := make(http.Header)
+	headers.Set(internal.SamsahaiAuthHeader, c.authToken)
+	ctx, err = twirp.WithHTTPRequestHeaders(ctx, headers)
+	if err != nil {
+		logger.Error(err, "cannot set request header")
+	}
+
+	rpcComp := &samsahairpc.ComponentUpgrade{
+		Name:      q.Spec.Name,
+		Namespace: q.Namespace,
+	}
+	if c.s2hClient != nil {
+		if _, err := c.s2hClient.SendUpdateStateQueueMetric(ctx, rpcComp); err != nil {
+			logger.Error(err, "cannot send updateQueueWithState queue metric")
+		}
+	}
+
 	comp.Status.UpdatedAt = &now
 	if err := c.client.Update(context.TODO(), comp); err != nil {
 		return reconcile.Result{}, err

internal/samsahai/exporter/metric.go

@@ -12,13 +12,18 @@ import (
 )
 
 type ActivePromotionMetricState string
+type QueueMetricState string
 
 const (
-	stateWaiting    ActivePromotionMetricState = "waiting"
-	stateDeploying  ActivePromotionMetricState = "deploying"
-	stateTesting    ActivePromotionMetricState = "testing"
-	statePromoting  ActivePromotionMetricState = "promoting"
-	stateDestroying ActivePromotionMetricState = "destroying"
+	stateWaiting        ActivePromotionMetricState = "waiting"
+	stateDeploying      ActivePromotionMetricState = "deploying"
+	stateTesting        ActivePromotionMetricState = "testing"
+	statePromoting      ActivePromotionMetricState = "promoting"
+	stateDestroying     ActivePromotionMetricState = "destroying"
+	queueStateWaiting   QueueMetricState           = "waiting"
+	queueStateDeploying QueueMetricState           = "deploying"
+	queueStateTesting   QueueMetricState           = "testing"
+	queueStateCleaning  QueueMetricState           = "cleaning"
 )
 
 var logger = s2hlog.S2HLog.WithName("exporter")
@@ -36,7 +41,7 @@ var HealthStatusMetric = prometheus.NewGaugeVec(prometheus.GaugeOpts{
 var QueueMetric = prometheus.NewGaugeVec(prometheus.GaugeOpts{
 	Name: "samsahai_queue",
 	Help: "Show components in queue",
-}, []string{"order", "teamName", "component", "version", "state", "no_of_processed"})
+}, []string{"teamName", "component", "version", "state", "order", "no_of_processed"})
 
 var ActivePromotionMetric = prometheus.NewGaugeVec(prometheus.GaugeOpts{
 	Name: "samsahai_active_promotion",
@@ -56,71 +61,38 @@ func SetTeamNameMetric(teamList *s2hv1beta1.TeamList) {
 	}
 }
 
+func SetHealthStatusMetric(version, gitCommit string, ts float64) {
+	HealthStatusMetric.WithLabelValues(
+		version,
+		gitCommit).Set(ts)
+}
+
 func SetQueueMetric(queue *s2hv1beta1.Queue) {
-	queueStateList := map[string]float64{"waiting": 0, "testing": 0, "finished": 0, "deploying": 0, "cleaning": 0}
+	var queueState QueueMetricState
 	switch queue.Status.State {
 	case s2hv1beta1.Waiting:
-		queueStateList["waiting"] = 1
-		for state, val := range queueStateList {
-			QueueMetric.WithLabelValues(
-				strconv.Itoa(queue.Spec.NoOfOrder),
-				queue.Spec.TeamName,
-				queue.Name,
-				queue.Spec.Version,
-				state,
-				strconv.Itoa(queue.Status.NoOfProcessed)).Set(val)
-		}
+		queueState = queueStateWaiting
 	case s2hv1beta1.Testing, s2hv1beta1.Collecting:
-		queueStateList["testing"] = 1
-		for state, val := range queueStateList {
-			QueueMetric.WithLabelValues(
-				strconv.Itoa(queue.Spec.NoOfOrder),
-				queue.Spec.TeamName,
-				queue.Name,
-				queue.Spec.Version,
-				state,
-				strconv.Itoa(queue.Status.NoOfProcessed)).Set(val)
-		}
-	case s2hv1beta1.Finished:
-		queueStateList["finished"] = 1
-		for state, val := range queueStateList {
-			QueueMetric.WithLabelValues(
-				strconv.Itoa(queue.Spec.NoOfOrder),
-				queue.Spec.TeamName,
-				queue.Name,
-				queue.Spec.Version,
-				state,
-				strconv.Itoa(queue.Status.NoOfProcessed)).Set(val)
-		}
+		queueState = queueStateTesting
 	case s2hv1beta1.DetectingImageMissing, s2hv1beta1.Creating:
-		queueStateList["deploying"] = 1
-		for state, val := range queueStateList {
-			QueueMetric.WithLabelValues(
-				strconv.Itoa(queue.Spec.NoOfOrder),
-				queue.Spec.TeamName,
-				queue.Name,
-				queue.Spec.Version,
-				state,
-				strconv.Itoa(queue.Status.NoOfProcessed)).Set(val)
-		}
-	case s2hv1beta1.CleaningBefore, s2hv1beta1.CleaningAfter:
-		queueStateList["cleaning"] = 1
-		for state, val := range queueStateList {
-			QueueMetric.WithLabelValues(
-				strconv.Itoa(queue.Spec.NoOfOrder),
-				queue.Spec.TeamName,
-				queue.Name,
-				queue.Spec.Version,
-				state,
-				strconv.Itoa(queue.Status.NoOfProcessed)).Set(val)
+		queueState = queueStateDeploying
+	case s2hv1beta1.CleaningBefore:
+		queueState = queueStateCleaning
+	case s2hv1beta1.CleaningAfter:
+		q, err := QueueMetric.CurryWith(prometheus.Labels{"component": queue.Name, "version": queue.Spec.Version})
+		if err != nil {
+			logger.Error(err, "cannot get finished queue metric")
 		}
+		q.Reset()
 	}
-}
 
-func SetHealthStatusMetric(version, gitCommit string, ts float64) {
-	HealthStatusMetric.WithLabelValues(
-		version,
-		gitCommit).Set(ts)
+	QueueMetric.WithLabelValues(
+		queue.Spec.TeamName,
+		queue.Name,
+		queue.Spec.Version,
+		string(queueState),
+		strconv.Itoa(queue.Spec.NoOfOrder),
+		strconv.Itoa(queue.Status.NoOfProcessed)).Set(float64(time.Now().Unix()))
 }
 
 func SetActivePromotionMetric(atpComp *s2hv1beta1.ActivePromotion) {

internal/samsahai/internal_process.go

@@ -259,16 +259,6 @@ func (c *controller) updateTeamDesiredComponent(updateInfo updateTeamDesiredComp
 		return err
 	}
 
-	// Add metric updateQueueMetric
-	queue := &s2hv1beta1.Queue{}
-	if err = c.client.Get(ctx, types.NamespacedName{
-		Name:      compName,
-		Namespace: compNs}, queue); err != nil {
-		logger.Error(err, "cannot get the queue")
-	} else {
-		exporter.SetQueueMetric(queue)
-	}
-
 	return nil
 }