feat[sc-100944]: Support upserting and deleting custom metrics (#191)

* sync custom metrics

Author: Siva M

pact/custom_metrics_test.go

@@ -13,7 +13,10 @@ import (
 	"github.com/pact-foundation/pact-go/dsl"
 	"github.com/replicatedhq/kotskinds/apis/kots/v1beta1"
 	"github.com/replicatedhq/replicated-sdk/pkg/handlers"
+	"github.com/replicatedhq/replicated-sdk/pkg/k8sutil"
 	"github.com/replicatedhq/replicated-sdk/pkg/store"
+	"github.com/replicatedhq/replicated-sdk/pkg/util"
+	"k8s.io/client-go/kubernetes/fake"
 )
 
 func TestSendCustomAppMetrics(t *testing.T) {
@@ -63,6 +66,11 @@ func TestSendCustomAppMetrics(t *testing.T) {
 				Status: http.StatusOK,
 			})
 	}
+	fakeClientSet := fake.NewSimpleClientset(
+		k8sutil.CreateTestDeployment(util.GetReplicatedDeploymentName(), "default", "1", map[string]string{"app": "replicated"}),
+		k8sutil.CreateTestReplicaSet("replicated-sdk-instance-replicaset", "default", "1"),
+		k8sutil.CreateTestPod("replicated-sdk-instance-pod", "default", "replicated-sdk-instance-replicaset", map[string]string{"app": "replicated"}),
+	)
 	t.Run("Send valid custom app metrics", func(t *testing.T) {
 		pactInteraction()
 
@@ -72,11 +80,13 @@ func TestSendCustomAppMetrics(t *testing.T) {
 			ReplicatedAppEndpoint: license.Spec.Endpoint,
 			ChannelID:             license.Spec.ChannelID,
 			ChannelSequence:       channelSequence,
+			Namespace:             "default",
 		}
 		store.InitInMemory(storeOptions)
 		defer store.SetStore(nil)
 
 		if err := pact.Verify(func() error {
+			handlers.SetTestClientSet(fakeClientSet)
 			handlers.SendCustomAppMetrics(clientWriter, clientRequest)
 			if clientWriter.Code != http.StatusOK {
 				return fmt.Errorf("expected status code %d but got %d", http.StatusOK, clientWriter.Code)

pkg/apiserver/server.go

@@ -66,7 +66,8 @@ func Start(params APIServerParams) {
 	r.HandleFunc("/api/v1/app/info", handlers.GetCurrentAppInfo).Methods("GET")
 	r.HandleFunc("/api/v1/app/updates", handlers.GetAppUpdates).Methods("GET")
 	r.HandleFunc("/api/v1/app/history", handlers.GetAppHistory).Methods("GET")
-	r.HandleFunc("/api/v1/app/custom-metrics", handlers.SendCustomAppMetrics).Methods("POST")
+	r.HandleFunc("/api/v1/app/custom-metrics", handlers.SendCustomAppMetrics).Methods("POST", "PATCH")
+	r.HandleFunc("/api/v1/app/custom-metrics/{key}", handlers.DeleteCustomAppMetricsKey).Methods("DELETE")
 	r.HandleFunc("/api/v1/app/instance-tags", handlers.SendAppInstanceTags).Methods("POST")
 
 	// integration

pkg/handlers/app.go

@@ -9,6 +9,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/gorilla/mux"
 	"github.com/pkg/errors"
 	appstatetypes "github.com/replicatedhq/replicated-sdk/pkg/appstate/types"
 	"github.com/replicatedhq/replicated-sdk/pkg/config"
@@ -18,15 +19,16 @@ import (
 	"github.com/replicatedhq/replicated-sdk/pkg/k8sutil"
 	sdklicense "github.com/replicatedhq/replicated-sdk/pkg/license"
 	"github.com/replicatedhq/replicated-sdk/pkg/logger"
+	"github.com/replicatedhq/replicated-sdk/pkg/meta"
+	"github.com/replicatedhq/replicated-sdk/pkg/meta/types"
 	"github.com/replicatedhq/replicated-sdk/pkg/report"
 	"github.com/replicatedhq/replicated-sdk/pkg/store"
-	"github.com/replicatedhq/replicated-sdk/pkg/tags"
-	"github.com/replicatedhq/replicated-sdk/pkg/tags/types"
 	"github.com/replicatedhq/replicated-sdk/pkg/upstream"
 	upstreamtypes "github.com/replicatedhq/replicated-sdk/pkg/upstream/types"
 	"github.com/replicatedhq/replicated-sdk/pkg/util"
 	helmrelease "helm.sh/helm/v3/pkg/release"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/scheme"
 )
 
@@ -352,6 +354,12 @@ func mockReleaseToAppRelease(mockRelease integrationtypes.MockRelease) AppReleas
 	return appRelease
 }
 
+var testClientSet kubernetes.Interface
+
+func SetTestClientSet(clientset kubernetes.Interface) {
+	testClientSet = clientset
+}
+
 func SendCustomAppMetrics(w http.ResponseWriter, r *http.Request) {
 	request := SendCustomAppMetricsRequest{}
 	if err := json.NewDecoder(r.Body).Decode(&request); err != nil {
@@ -366,22 +374,60 @@ func SendCustomAppMetrics(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	var clientset kubernetes.Interface
+	if testClientSet != nil {
+		clientset = testClientSet
+	} else {
+		var err error
+		clientset, err = k8sutil.GetClientset()
+		if err != nil {
+			logger.Error(errors.Wrap(err, "failed to get clientset"))
+			w.WriteHeader(http.StatusInternalServerError)
+			return
+		}
+	}
+
+	overwrite := true
+	if r.Method == http.MethodPatch {
+		overwrite = false
+	}
+
+	if err := report.SendCustomAppMetrics(clientset, store.GetStore(), request.Data, overwrite); err != nil {
+		logger.Error(errors.Wrap(err, "set application data"))
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
+
+	JSON(w, http.StatusOK, "")
+}
+
+func DeleteCustomAppMetricsKey(w http.ResponseWriter, r *http.Request) {
+	key, ok := mux.Vars(r)["key"]
+
+	if !ok || key == "" {
+		w.WriteHeader(http.StatusBadRequest)
+		fmt.Fprintf(w, "key cannot be empty")
+		logger.Errorf("cannot delete custom metrics key - key cannot be empty")
+		return
+	}
+
 	clientset, err := k8sutil.GetClientset()
 	if err != nil {
 		logger.Error(errors.Wrap(err, "failed to get clientset"))
 		w.WriteHeader(http.StatusInternalServerError)
 		return
 	}
 
-	if err := report.SendCustomAppMetrics(clientset, store.GetStore(), request.Data); err != nil {
-		logger.Error(errors.Wrap(err, "set application data"))
+	data := map[string]interface{}{key: nil}
+
+	if err := report.SendCustomAppMetrics(clientset, store.GetStore(), data, false); err != nil {
+		logger.Error(errors.Wrapf(err, "failed to delete custom merics key: %s", key))
 		w.WriteHeader(http.StatusBadRequest)
 		return
 	}
 
-	JSON(w, http.StatusOK, "")
+	JSON(w, http.StatusNoContent, "")
 }
-
 func validateCustomAppMetricsData(data CustomAppMetricsData) error {
 	if len(data) == 0 {
 		return errors.New("no data provided")
@@ -429,7 +475,7 @@ func SendAppInstanceTags(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	if err := tags.Save(r.Context(), clientset, store.GetStore().GetNamespace(), request.Data); err != nil {
+	if err := meta.SaveInstanceTag(r.Context(), clientset, store.GetStore().GetNamespace(), request.Data); err != nil {
 		logger.Errorf("failed to save instance tags: %v", err)
 		w.WriteHeader(http.StatusInternalServerError)
 		return

pkg/k8sutil/fake.go

@@ -14,6 +14,7 @@ func CreateTestDeployment(name string, namespace string, revision string, matchL
 			Annotations: map[string]string{
 				"deployment.kubernetes.io/revision": revision,
 			},
+			UID: "test-deployment-uid",
 		},
 		Spec: appsv1.DeploymentSpec{
 			Selector: &metav1.LabelSelector{
@@ -31,13 +32,15 @@ func CreateTestReplicaSet(name string, namespace string, revision string) *appsv
 			Annotations: map[string]string{
 				"deployment.kubernetes.io/revision": revision,
 			},
+			UID: "test-deployment-uid",
 		},
 	}
 }
 
 func CreateTestPod(name string, namespace string, replicaSetName string, labels map[string]string) *corev1.Pod {
 	return &corev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{
+			UID:       "test-deployment-uid",
 			Name:      name,
 			Namespace: namespace,
 			Labels:    labels,

pkg/meta/instance_tags.go

@@ -0,0 +1,28 @@
+package meta
+
+import (
+	"context"
+
+	"github.com/pkg/errors"
+	"github.com/replicatedhq/replicated-sdk/pkg/meta/types"
+	"k8s.io/client-go/kubernetes"
+)
+
+const (
+	instanceTagSecretKey replicatedMetadataSecretKey = "instance-tag-data"
+)
+
+func SaveInstanceTag(ctx context.Context, clientset kubernetes.Interface, namespace string, tdata types.InstanceTagData) error {
+	return save(ctx, clientset, namespace, instanceTagSecretKey, tdata)
+}
+
+func GetInstanceTag(ctx context.Context, clientset kubernetes.Interface, namespace string) (*types.InstanceTagData, error) {
+	t := types.InstanceTagData{}
+
+	if err := get(ctx, clientset, namespace, instanceTagSecretKey, &t); err != nil {
+		return nil, errors.Wrapf(err, "failed to get instance tag data")
+	}
+
+	return &t, nil
+
+}

pkg/meta/latest_custom_metrics.go

@@ -0,0 +1,59 @@
+package meta
+
+import (
+	"context"
+	"maps"
+
+	"github.com/pkg/errors"
+	"k8s.io/client-go/kubernetes"
+)
+
+const (
+	customMetricsSecretKey replicatedMetadataSecretKey = "latest-custom-metrics"
+)
+
+func SyncCustomAppMetrics(ctx context.Context, clientset kubernetes.Interface, namespace string, inboundMetrics map[string]interface{}, overwrite bool) (map[string]interface{}, error) {
+	existing := map[string]interface{}{}
+
+	err := get(ctx, clientset, namespace, customMetricsSecretKey, &existing)
+	if err != nil && errors.Cause(err) != ErrReplicatedMetadataSecretNotFound {
+		return nil, errors.Wrapf(err, "failed to get custom metrics data")
+	}
+
+	modified := mergeCustomAppMetrics(existing, inboundMetrics, overwrite)
+
+	if err := save(ctx, clientset, namespace, customMetricsSecretKey, modified); err != nil {
+		return nil, errors.Wrap(err, "failed to save custom metrics")
+	}
+
+	return modified, nil
+}
+
+func mergeCustomAppMetrics(existingMetrics map[string]interface{}, inboundMetrics map[string]interface{}, overwrite bool) map[string]interface{} {
+	if existingMetrics == nil {
+		existingMetrics = map[string]interface{}{}
+	}
+
+	if inboundMetrics == nil {
+		inboundMetrics = map[string]interface{}{}
+	}
+
+	if overwrite {
+		return inboundMetrics
+	}
+
+	if len(inboundMetrics) == 0 || maps.Equal(existingMetrics, inboundMetrics) {
+		return existingMetrics
+	}
+
+	for k, v := range inboundMetrics {
+		if v == nil {
+			delete(existingMetrics, k)
+			continue
+		}
+
+		existingMetrics[k] = v
+	}
+
+	return existingMetrics
+}

pkg/meta/latest_custom_metrics_test.go

@@ -0,0 +1,83 @@
+package meta
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func Test_mergeCustomAppMetrics(tst *testing.T) {
+	tests := []struct {
+		name            string
+		existingMetrics map[string]interface{}
+		inboundMetrics  map[string]interface{}
+		overwrite       bool
+		assertFn        func(*testing.T, map[string]interface{})
+	}{
+		{
+			name:            "should return empty if both are empty",
+			existingMetrics: map[string]interface{}{},
+			inboundMetrics:  map[string]interface{}{},
+			overwrite:       false,
+			assertFn: func(t *testing.T, actual map[string]interface{}) {
+				assert.NotNil(t, actual)
+				assert.Empty(t, actual)
+			},
+		},
+		{
+			name:            "should tolerate nil value on existingMetrics",
+			existingMetrics: nil,
+			inboundMetrics:  map[string]interface{}{"numProjects": 10},
+			overwrite:       false,
+			assertFn: func(t *testing.T, actual map[string]interface{}) {
+				expected := map[string]interface{}{"numProjects": 10}
+				assert.Equal(t, expected, actual)
+			},
+		},
+		{
+			name:            "should tolerate nil value on inboundMetrics",
+			existingMetrics: map[string]interface{}{"numProjects": 10},
+			inboundMetrics:  nil,
+			overwrite:       false,
+			assertFn: func(t *testing.T, actual map[string]interface{}) {
+				expected := map[string]interface{}{"numProjects": 10}
+				assert.Equal(t, expected, actual)
+			},
+		},
+		{
+			name:            "should return inboundMetrics when overwrite is true",
+			existingMetrics: map[string]interface{}{"numProjects": 10, "token": "1234"},
+			inboundMetrics:  map[string]interface{}{"newProjects": 100, "newToken": 10},
+			overwrite:       true, // overwrites existing metric data with inbound metrics data
+			assertFn: func(t *testing.T, actual map[string]interface{}) {
+				expected := map[string]interface{}{"newProjects": 100, "newToken": 10}
+				assert.Equal(t, expected, actual)
+			},
+		},
+		{
+			name:            "should return merged data when overwrite is false",
+			existingMetrics: map[string]interface{}{"numProjects": 10, "token": "1234"},
+			inboundMetrics:  map[string]interface{}{"numProjects": 66666, "numPeople": 100},
+			overwrite:       false,
+			assertFn: func(t *testing.T, actual map[string]interface{}) {
+				expected := map[string]interface{}{"numPeople": 100, "numProjects": 66666, "token": "1234"}
+				assert.Equal(t, expected, actual)
+			},
+		},
+		{
+			name:            "should delete existing metric key when the corresponding inboundMetrics value is nil",
+			existingMetrics: map[string]interface{}{"numProjects": 10, "token": "1234"},
+			inboundMetrics:  map[string]interface{}{"numProjects": nil}, // delete numProjects
+			overwrite:       false,
+			assertFn: func(t *testing.T, actual map[string]interface{}) {
+				expected := map[string]interface{}{"token": "1234"}
+				assert.Equal(t, expected, actual)
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		m := mergeCustomAppMetrics(tt.existingMetrics, tt.inboundMetrics, tt.overwrite)
+		tt.assertFn(tst, m)
+	}
+}