Merge pull request #15 from rizalgowandy/arwego/feat/v2

Add Telemetry Interceptor

interceptor/distributed_lock_test.go

@@ -0,0 +1,30 @@
+package interceptor
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestDistributedLock(t *testing.T) {
+	type args struct {
+		serviceName string
+		mode        string
+		dl          DistributedLockItf
+		sc          SlackClientItf
+	}
+	tests := []struct {
+		name string
+		args args
+	}{
+		{
+			name: "Success",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := DistributedLock(tt.args.serviceName, tt.args.mode, tt.args.dl, tt.args.sc)
+			assert.NotNil(t, got)
+		})
+	}
+}

interceptor/slack_test.go

@@ -0,0 +1,29 @@
+package interceptor
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNotifySlack(t *testing.T) {
+	type args struct {
+		serviceName string
+		mode        string
+		sc          SlackClientItf
+	}
+	tests := []struct {
+		name string
+		args args
+	}{
+		{
+			name: "Success",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := NotifySlack(tt.args.serviceName, tt.args.mode, tt.args.sc)
+			assert.NotNil(t, got)
+		})
+	}
+}

interceptor/telemetry.go

@@ -0,0 +1,74 @@
+package interceptor
+
+import (
+	"context"
+	"time"
+
+	"github.com/rizalgowandy/cronx"
+	"github.com/rizalgowandy/gdk/pkg/errorx/v2"
+	"github.com/rizalgowandy/gdk/pkg/tags"
+)
+
+const (
+	MetricCronPerformance = "cron_performance"
+)
+
+type TelemetryClientItf interface {
+	Histogram(metricName string, value float64, tags map[string]string)
+}
+
+// Telemetry is a middleware that push the latency of a process.
+func Telemetry(metric TelemetryClientItf) cronx.Interceptor {
+	const dividerNStoMS = 1e6
+
+	return func(ctx context.Context, job *cronx.Job, handler cronx.Handler) error {
+		start := time.Now()
+		err := handler(ctx, job)
+
+		// Publish current metric.
+		go func() {
+			metric.Histogram(
+				MetricCronPerformance,
+				float64(time.Since(start).Nanoseconds()/dividerNStoMS),
+				DefaultKV(job.Name, err),
+			)
+		}()
+
+		return err
+	}
+}
+
+// DefaultKV returns a standardized service tracking.
+func DefaultKV(method string, err error) map[string]string {
+	trackingTags := map[string]string{
+		tags.Method:       method,
+		tags.MetricStatus: string(errorx.MetricStatusSuccess),
+	}
+
+	// Set status as error.
+	if err != nil {
+		trackingTags[tags.Error] = err.Error()
+		trackingTags[tags.MetricStatus] = string(errorx.MetricStatusErr)
+
+		// If error is our custom error, add additional tags.
+		if e, ok := err.(*errorx.Error); ok {
+			if e.MetricStatus != "" {
+				trackingTags[tags.MetricStatus] = string(e.MetricStatus)
+			}
+			if e.Code != "" {
+				trackingTags[tags.Code] = string(e.Code)
+			}
+			if e.Line != "" {
+				trackingTags[tags.ErrorLine] = string(e.Line)
+			}
+			if e.Message != "" {
+				trackingTags[tags.Message] = string(e.Message)
+			}
+			if len(e.OpTraces) > 0 {
+				trackingTags[tags.Ops] = string(e.OpTraces[len(e.OpTraces)-1])
+			}
+		}
+	}
+
+	return trackingTags
+}

interceptor/telemetry_test.go

@@ -0,0 +1,27 @@
+package interceptor
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestTelemetry(t *testing.T) {
+	type args struct {
+		metric TelemetryClientItf
+	}
+	tests := []struct {
+		name string
+		args args
+	}{
+		{
+			name: "Success",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := Telemetry(tt.args.metric)
+			assert.NotNil(t, got)
+		})
+	}
+}