Add Linkerd observer implementation

pkg/metrics/factory.go

@@ -31,6 +31,10 @@ func (factory Factory) Observer() Interface {
 		return &NginxObserver{
 			client: factory.Client,
 		}
+	case factory.MeshProvider == "smi:linkerd":
+		return &LinkerdObserver{
+			client: factory.Client,
+		}
 	default:
 		return &IstioObserver{
 			client: factory.Client,

pkg/metrics/linkerd.go

@@ -0,0 +1,73 @@
+package metrics
+
+import (
+	"time"
+)
+
+var linkerdQueries = map[string]string{
+	"request-success-rate": `
+	sum(
+		rate(
+			response_total{
+				namespace="{{ .Namespace }}",
+				dst_deployment=~"{{ .Name }}",
+				classification="failure"
+			}[{{ .Interval }}]
+		)
+	) 
+	/ 
+	sum(
+		rate(
+			response_total{
+				namespace="{{ .Namespace }}",
+				dst_deployment=~"{{ .Name }}"
+			}[{{ .Interval }}]
+		)
+	) 
+	* 100`,
+	"request-duration": `
+	histogram_quantile(
+		0.99,
+		sum(
+			rate(
+				response_latency_ms_bucket{
+					namespace="{{ .Namespace }}",
+					dst_deployment=~"{{ .Name }}"
+				}[{{ .Interval }}]
+			)
+		) by (le)
+	)`,
+}
+
+type LinkerdObserver struct {
+	client *PrometheusClient
+}
+
+func (ob *LinkerdObserver) GetRequestSuccessRate(name string, namespace string, interval string) (float64, error) {
+	query, err := ob.client.RenderQuery(name, namespace, interval, linkerdQueries["request-success-rate"])
+	if err != nil {
+		return 0, err
+	}
+
+	value, err := ob.client.RunQuery(query)
+	if err != nil {
+		return 0, err
+	}
+
+	return value, nil
+}
+
+func (ob *LinkerdObserver) GetRequestDuration(name string, namespace string, interval string) (time.Duration, error) {
+	query, err := ob.client.RenderQuery(name, namespace, interval, linkerdQueries["request-duration"])
+	if err != nil {
+		return 0, err
+	}
+
+	value, err := ob.client.RunQuery(query)
+	if err != nil {
+		return 0, err
+	}
+
+	ms := time.Duration(int64(value)) * time.Millisecond
+	return ms, nil
+}

pkg/metrics/linkerd_test.go

@@ -0,0 +1,74 @@
+package metrics
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+)
+
+func TestLinkerdObserver_GetRequestSuccessRate(t *testing.T) {
+	expected := `sum(rate(response_total{namespace="default",dst_deployment=~"podinfo",classification="failure"}[1m]))/sum(rate(response_total{namespace="default",dst_deployment=~"podinfo"}[1m]))*100`
+
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		promql := r.URL.Query()["query"][0]
+		if promql != expected {
+			t.Errorf("\nGot %s \nWanted %s", promql, expected)
+		}
+
+		json := `{"status":"success","data":{"resultType":"vector","result":[{"metric":{},"value":[1,"100"]}]}}`
+		w.Write([]byte(json))
+	}))
+	defer ts.Close()
+
+	client, err := NewPrometheusClient(ts.URL, time.Second)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	observer := &LinkerdObserver{
+		client: client,
+	}
+
+	val, err := observer.GetRequestSuccessRate("podinfo", "default", "1m")
+	if err != nil {
+		t.Fatal(err.Error())
+	}
+
+	if val != 100 {
+		t.Errorf("Got %v wanted %v", val, 100)
+	}
+}
+
+func TestLinkerdObserver_GetRequestDuration(t *testing.T) {
+	expected := `histogram_quantile(0.99,sum(rate(response_latency_ms_bucket{namespace="default",dst_deployment=~"podinfo"}[1m]))by(le))`
+
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		promql := r.URL.Query()["query"][0]
+		if promql != expected {
+			t.Errorf("\nGot %s \nWanted %s", promql, expected)
+		}
+
+		json := `{"status":"success","data":{"resultType":"vector","result":[{"metric":{},"value":[1,"100"]}]}}`
+		w.Write([]byte(json))
+	}))
+	defer ts.Close()
+
+	client, err := NewPrometheusClient(ts.URL, time.Second)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	observer := &LinkerdObserver{
+		client: client,
+	}
+
+	val, err := observer.GetRequestDuration("podinfo", "default", "1m")
+	if err != nil {
+		t.Fatal(err.Error())
+	}
+
+	if val != 100*time.Millisecond {
+		t.Errorf("Got %v wanted %v", val, 100*time.Millisecond)
+	}
+}