feat: add metrics for res status change (#126)

* poc

* feat: add metrics for res status change

* chore: remove unused variables

* refactor: otel setup

* chore: clear opencencus

* fix: add runtime metrics

* test: fix

* feat: add mod file

---------

Co-authored-by: Ishan Arya <[email protected]>

Author: ishanarya0

cli/serve.go

@@ -64,7 +64,7 @@ func StartServer(ctx context.Context, cfg Config, migrate, spawnWorker bool) err
 
 	store := setupStorage(cfg.PGConnStr, cfg.Syncer, cfg.Service)
 	moduleService := module.NewService(setupRegistry(), store)
-	resourceService := core.New(store, moduleService, time.Now, cfg.Syncer.SyncBackoffInterval, cfg.Syncer.MaxRetries)
+	resourceService := core.New(store, moduleService, time.Now, cfg.Syncer.SyncBackoffInterval, cfg.Syncer.MaxRetries, cfg.Telemetry.ServiceName)
 
 	if migrate {
 		if migrateErr := runMigrations(ctx, cfg); migrateErr != nil {

cli/worker.go

@@ -58,7 +58,7 @@ func cmdWorker() *cobra.Command {
 func StartWorkers(ctx context.Context, cfg Config) error {
 	store := setupStorage(cfg.PGConnStr, cfg.Syncer, cfg.Service)
 	moduleService := module.NewService(setupRegistry(), store)
-	resourceService := core.New(store, moduleService, time.Now, cfg.Syncer.SyncBackoffInterval, cfg.Syncer.MaxRetries)
+	resourceService := core.New(store, moduleService, time.Now, cfg.Syncer.SyncBackoffInterval, cfg.Syncer.MaxRetries, cfg.Telemetry.ServiceName)
 
 	eg := &errgroup.Group{}
 	spawnWorkers(ctx, resourceService, cfg.Syncer.Workers, cfg.Syncer.SyncInterval, eg)

core/core.go

@@ -18,6 +18,7 @@ type Service struct {
 	moduleSvc      ModuleService
 	syncBackoff    time.Duration
 	maxSyncRetries int
+	serviceName    string
 }
 
 type ModuleService interface {
@@ -27,7 +28,7 @@ type ModuleService interface {
 	GetOutput(ctx context.Context, res module.ExpandedResource) (json.RawMessage, error)
 }
 
-func New(repo resource.Store, moduleSvc ModuleService, clockFn func() time.Time, syncBackoffInterval time.Duration, maxRetries int) *Service {
+func New(repo resource.Store, moduleSvc ModuleService, clockFn func() time.Time, syncBackoffInterval time.Duration, maxRetries int, serviceName string) *Service {
 	if clockFn == nil {
 		clockFn = time.Now
 	}
@@ -38,6 +39,7 @@ func New(repo resource.Store, moduleSvc ModuleService, clockFn func() time.Time,
 		syncBackoff:    syncBackoffInterval,
 		maxSyncRetries: maxRetries,
 		moduleSvc:      moduleSvc,
+		serviceName:    serviceName,
 	}
 }
 

core/core_test.go

@@ -25,6 +25,6 @@ var (
 
 func TestNew(t *testing.T) {
 	t.Parallel()
-	s := core.New(&mocks.ResourceStore{}, &mocks.ModuleService{}, deadClock, defaultSyncBackoff, defaultMaxRetries)
+	s := core.New(&mocks.ResourceStore{}, &mocks.ModuleService{}, deadClock, defaultSyncBackoff, defaultMaxRetries, serviceName)
 	assert.NotNil(t, s)
 }

core/read_test.go

@@ -17,6 +17,7 @@ import (
 const (
 	defaultMaxRetries  = 5
 	defaultSyncBackoff = 5 * time.Second
+	serviceName        = "test-service"
 )
 
 func TestService_GetResource(t *testing.T) {
@@ -38,7 +39,7 @@ func TestService_GetResource(t *testing.T) {
 					GetByURN(mock.Anything, mock.Anything).
 					Return(nil, errors.ErrNotFound).
 					Once()
-				return core.New(repo, nil, nil, defaultSyncBackoff, defaultMaxRetries)
+				return core.New(repo, nil, nil, defaultSyncBackoff, defaultMaxRetries, serviceName)
 			},
 			urn:     "foo:bar:baz",
 			wantErr: errors.ErrNotFound,
@@ -58,7 +59,7 @@ func TestService_GetResource(t *testing.T) {
 					Return(nil, nil).
 					Once()
 
-				return core.New(repo, mod, deadClock, defaultSyncBackoff, defaultMaxRetries)
+				return core.New(repo, mod, deadClock, defaultSyncBackoff, defaultMaxRetries, serviceName)
 			},
 			urn:     "foo:bar:baz",
 			want:    &sampleResource,
@@ -105,7 +106,7 @@ func TestService_ListResources(t *testing.T) {
 					List(mock.Anything, mock.Anything, false).
 					Return(nil, nil).
 					Once()
-				return core.New(repo, nil, deadClock, defaultSyncBackoff, defaultMaxRetries)
+				return core.New(repo, nil, deadClock, defaultSyncBackoff, defaultMaxRetries, serviceName)
 			},
 			want:    resource.PagedResource{},
 			wantErr: nil,
@@ -119,7 +120,7 @@ func TestService_ListResources(t *testing.T) {
 					List(mock.Anything, mock.Anything, false).
 					Return(nil, errStoreFailure).
 					Once()
-				return core.New(repo, nil, deadClock, defaultSyncBackoff, defaultMaxRetries)
+				return core.New(repo, nil, deadClock, defaultSyncBackoff, defaultMaxRetries, serviceName)
 			},
 			want:    resource.PagedResource{},
 			wantErr: errors.ErrInternal,
@@ -133,7 +134,7 @@ func TestService_ListResources(t *testing.T) {
 					List(mock.Anything, mock.Anything, false).
 					Return([]resource.Resource{sampleResource}, nil).
 					Once()
-				return core.New(repo, nil, deadClock, defaultSyncBackoff, defaultMaxRetries)
+				return core.New(repo, nil, deadClock, defaultSyncBackoff, defaultMaxRetries, serviceName)
 			},
 			want: resource.PagedResource{
 				Count:     1,

core/sync.go

@@ -2,13 +2,26 @@ package core
 
 import (
 	"context"
+	"fmt"
 	"time"
 
+	"go.opentelemetry.io/otel/metric"
 	"go.uber.org/zap"
 	"golang.org/x/sync/errgroup"
 
 	"github.com/goto/entropy/core/resource"
 	"github.com/goto/entropy/pkg/errors"
+	"github.com/goto/entropy/pkg/telemetry"
+	"go.opentelemetry.io/otel/attribute"
+)
+
+type SyncStatus string
+
+const (
+	PendingCounter   SyncStatus = "pending"
+	CompletedCounter SyncStatus = "completed"
+	ErrorCounter     SyncStatus = "error"
+	RetryCounter     SyncStatus = "retry"
 )
 
 // RunSyncer runs the syncer thread that keeps performing resource-sync at
@@ -44,6 +57,20 @@ func (svc *Service) handleSync(ctx context.Context, res resource.Resource) (*res
 		zap.String("last_err", res.State.SyncResult.LastError),
 	)
 
+	meter := telemetry.GetMeter(svc.serviceName)
+	retryCounter, err := setupCounter(meter, RetryCounter)
+	if err != nil {
+		return nil, err
+	}
+	errorCounter, err := setupCounter(meter, ErrorCounter)
+	if err != nil {
+		return nil, err
+	}
+	completedCounter, err := setupCounter(meter, CompletedCounter)
+	if err != nil {
+		return nil, err
+	}
+
 	modSpec, err := svc.generateModuleSpec(ctx, res)
 	if err != nil {
 		logEntry.Error("SyncOne() failed", zap.Error(err))
@@ -56,16 +83,26 @@ func (svc *Service) handleSync(ctx context.Context, res resource.Resource) (*res
 
 		res.State.SyncResult.LastError = err.Error()
 		res.State.SyncResult.Retries++
+
+		// Increment the retry counter.
+		retryCounter.Add(context.Background(), 1, metric.WithAttributes(attribute.String("resource", res.URN)))
+
 		if errors.Is(err, errors.ErrInvalid) {
 			// ErrInvalid is expected to be returned when config is invalid.
 			// There is no point in retrying in this case.
 			res.State.Status = resource.StatusError
 			res.State.NextSyncAt = nil
+
+			// Increment the error counter.
+			errorCounter.Add(context.Background(), 1)
 		} else if svc.maxSyncRetries > 0 && res.State.SyncResult.Retries >= svc.maxSyncRetries {
 			// Some other error occurred and no more retries remaining.
 			// move the resource to failure state.
 			res.State.Status = resource.StatusError
 			res.State.NextSyncAt = nil
+
+			// Increment the error counter.
+			errorCounter.Add(context.Background(), 1)
 		} else {
 			// Some other error occurred and we still have remaining retries.
 			// need to backoff and retry in some time.
@@ -78,6 +115,9 @@ func (svc *Service) handleSync(ctx context.Context, res resource.Resource) (*res
 		res.UpdatedAt = svc.clock()
 		res.State = *newState
 
+		// Increment the completed counter.
+		completedCounter.Add(context.Background(), 1)
+
 		logEntry.Info("SyncOne() finished",
 			zap.String("final_status", res.State.Status),
 			zap.Timep("next_sync", res.State.NextSyncAt),
@@ -86,3 +126,11 @@ func (svc *Service) handleSync(ctx context.Context, res resource.Resource) (*res
 
 	return &res, nil
 }
+
+func setupCounter(meter metric.Meter, countername SyncStatus) (metric.Int64Counter, error) {
+	return meter.Int64Counter(
+		fmt.Sprintf("%s_counter", countername),
+		metric.WithDescription(fmt.Sprintf("Total number of %s performed", countername)),
+		metric.WithUnit("1"),
+	)
+}

core/write.go

@@ -7,6 +7,7 @@ import (
 	"github.com/goto/entropy/core/module"
 	"github.com/goto/entropy/core/resource"
 	"github.com/goto/entropy/pkg/errors"
+	"github.com/goto/entropy/pkg/telemetry"
 )
 
 type Options struct {
@@ -101,6 +102,16 @@ func (svc *Service) execAction(ctx context.Context, res resource.Resource, act m
 			return nil, err
 		}
 	}
+
+	meter := telemetry.GetMeter(svc.serviceName)
+	pendingCounter, err := setupCounter(meter, PendingCounter)
+	if err != nil {
+		return nil, err
+	}
+
+	// Increment the pending counter.
+	pendingCounter.Add(context.Background(), 1)
+
 	return planned, nil
 }