feat: Using secret to attach identity to pod for observability publis…

…her (#607)

<!--  Thanks for sending a pull request!  Here are some tips for you:

1. Run unit tests and ensure that they are passing
2. If your change introduces any API changes, make sure to update the
e2e tests
3. Make sure documentation is updated for your PR!

-->
# Description
Since we are trying to move away from cloud-specific feature, hence as
first step in observability publisher is to use secret for propagate
identity instead of using existing workload identity way

# Modifications
Add `GOOGLE_APPLICATION_CREDENTIALS` for propagate google identity to
pod
# Tests
<!-- Besides the existing / updated automated tests, what specific
scenarios should be tested? Consider the backward compatibility of the
changes, whether corner cases are covered, etc. Please describe the
tests and check the ones that have been completed. Eg:
- [x] Deploying new and existing standard models
- [ ] Deploying PyFunc models
-->

# Checklist
- [ ] Added PR label
- [ ] Added unit test, integration, and/or e2e tests
- [ ] Tested locally
- [ ] Updated documentation
- [ ] Update Swagger spec if the PR introduce API changes
- [ ] Regenerated Golang and Python client if the PR introduces API
changes

# Release Notes
<!--
Does this PR introduce a user-facing change?
If no, just write "NONE" in the release-note block below.
If yes, a release note is required. Enter your extended release note in
the block below.
If the PR requires additional action from users switching to the new
release, include the string "action required".

For more information about release notes, see kubernetes' guide here:
http://git.k8s.io/community/contributors/guide/release-notes.md
-->

```release-note

```

api/config/observability.go

@@ -4,16 +4,16 @@ import "time"
 
 // ObservabilityPublisher
 type ObservabilityPublisher struct {
-	ArizeSink          ArizeSink
-	BigQuerySink       BigQuerySink
-	KafkaConsumer      KafkaConsumer
-	ImageName          string
-	DefaultResources   ResourceRequestsLimits
-	EnvironmentName    string
-	Replicas           int32
-	TargetNamespace    string
-	ServiceAccountName string
-	DeploymentTimeout  time.Duration `default:"30m"`
+	ArizeSink                ArizeSink
+	BigQuerySink             BigQuerySink
+	KafkaConsumer            KafkaConsumer
+	ImageName                string
+	DefaultResources         ResourceRequestsLimits
+	EnvironmentName          string
+	Replicas                 int32
+	TargetNamespace          string
+	DeploymentTimeout        time.Duration `default:"30m"`
+	ServiceAccountSecretName string
 }
 
 // KafkaConsumer

api/pkg/observability/deployment/deployment.go

@@ -346,6 +346,52 @@ func (c *deployer) createDeploymentSpec(data *models.WorkerData, secretName stri
 	labels := c.getLabels(data)
 
 	cfgVolName := "config-volume"
+	workerContainer := "worker"
+	podSpec := corev1.PodSpec{
+		Containers: []corev1.Container{
+			{
+				Name:  workerContainer,
+				Image: c.consumerConfig.ImageName,
+				Command: []string{
+					"python",
+					"-m",
+					"publisher",
+					"+environment=config",
+				},
+				ImagePullPolicy: corev1.PullIfNotPresent,
+
+				Resources: corev1.ResourceRequirements{
+					Requests: c.resourceRequest,
+					Limits:   c.resourceLimit,
+				},
+				VolumeMounts: []corev1.VolumeMount{
+					{
+						Name:      cfgVolName,
+						MountPath: "/mlobs/observation-publisher/conf/environment",
+						ReadOnly:  true,
+					},
+				},
+				Ports: []corev1.ContainerPort{
+					{
+						Name:          "prom-metric",
+						ContainerPort: 8000,
+						Protocol:      corev1.ProtocolTCP,
+					},
+				},
+			},
+		},
+		Volumes: []corev1.Volume{
+			{
+				Name: cfgVolName,
+				VolumeSource: corev1.VolumeSource{
+					Secret: &corev1.SecretVolumeSource{
+						SecretName: secretName,
+					},
+				},
+			},
+		},
+	}
+	podSpecWithIdentity := enrichIdentityToPod(podSpec, c.consumerConfig.ServiceAccountSecretName, []string{workerContainer})
 	return &appsv1.Deployment{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      c.getDeploymentName(data),
@@ -369,52 +415,7 @@ func (c *deployer) createDeploymentSpec(data *models.WorkerData, secretName stri
 						PublisherRevisionAnnotationKey: strconv.Itoa(data.Revision),
 					},
 				},
-
-				Spec: corev1.PodSpec{
-					Containers: []corev1.Container{
-						{
-							Name:  "worker",
-							Image: c.consumerConfig.ImageName,
-							Command: []string{
-								"python",
-								"-m",
-								"publisher",
-								"+environment=config",
-							},
-							ImagePullPolicy: corev1.PullIfNotPresent,
-
-							Resources: corev1.ResourceRequirements{
-								Requests: c.resourceRequest,
-								Limits:   c.resourceLimit,
-							},
-							VolumeMounts: []corev1.VolumeMount{
-								{
-									Name:      cfgVolName,
-									MountPath: "/mlobs/observation-publisher/conf/environment",
-									ReadOnly:  true,
-								},
-							},
-							Ports: []corev1.ContainerPort{
-								{
-									Name:          "prom-metric",
-									ContainerPort: 8000,
-									Protocol:      corev1.ProtocolTCP,
-								},
-							},
-						},
-					},
-					Volumes: []corev1.Volume{
-						{
-							Name: cfgVolName,
-							VolumeSource: corev1.VolumeSource{
-								Secret: &corev1.SecretVolumeSource{
-									SecretName: secretName,
-								},
-							},
-						},
-					},
-					ServiceAccountName: c.consumerConfig.ServiceAccountName,
-				},
+				Spec: podSpecWithIdentity,
 			},
 		},
 	}, nil

api/pkg/observability/deployment/deployment_test.go

@@ -48,11 +48,11 @@ const (
 	ready        deploymentStatus = "ready"
 	timeoutError deploymentStatus = "timeout_error"
 
-	namespace          = "caraml-observability"
-	serviceAccountName = "caraml-observability-sa"
+	namespace                = "caraml-observability"
+	serviceAccountSecretName = "caraml-observability-sa-secret"
 )
 
-func createDeploymentSpec(data *models.WorkerData, resourceRequest corev1.ResourceList, resourceLimit corev1.ResourceList, imageName string) *appsv1.Deployment {
+func createDeploymentSpec(data *models.WorkerData, resourceRequest corev1.ResourceList, resourceLimit corev1.ResourceList, imageName string, serviceAccountSecretName string) *appsv1.Deployment {
 	labels := data.Metadata.ToLabel()
 	labels[appLabelKey] = data.Metadata.App
 	numReplicas := int32(2)
@@ -80,7 +80,6 @@ func createDeploymentSpec(data *models.WorkerData, resourceRequest corev1.Resour
 					},
 				},
 				Spec: corev1.PodSpec{
-					ServiceAccountName: serviceAccountName,
 					Containers: []corev1.Container{
 						{
 							Name:  "worker",
@@ -103,6 +102,11 @@ func createDeploymentSpec(data *models.WorkerData, resourceRequest corev1.Resour
 									MountPath: "/mlobs/observation-publisher/conf/environment",
 									ReadOnly:  true,
 								},
+								{
+									Name:      "iam-secret",
+									MountPath: fmt.Sprintf("/iam/%s", serviceAccountSecretName),
+									ReadOnly:  true,
+								},
 							},
 							Ports: []corev1.ContainerPort{
 								{
@@ -111,6 +115,12 @@ func createDeploymentSpec(data *models.WorkerData, resourceRequest corev1.Resour
 									Protocol:      corev1.ProtocolTCP,
 								},
 							},
+							Env: []corev1.EnvVar{
+								{
+									Name:  "GOOGLE_APPLICATION_CREDENTIALS",
+									Value: fmt.Sprintf("/iam/%s/service-account.json", serviceAccountSecretName),
+								},
+							},
 						},
 					},
 					Volumes: []corev1.Volume{
@@ -122,6 +132,14 @@ func createDeploymentSpec(data *models.WorkerData, resourceRequest corev1.Resour
 								},
 							},
 						},
+						{
+							Name: "iam-secret",
+							VolumeSource: corev1.VolumeSource{
+								Secret: &corev1.SecretVolumeSource{
+									SecretName: serviceAccountSecretName,
+								},
+							},
+						},
 					},
 				},
 			},
@@ -213,11 +231,11 @@ func Test_deployer_Deploy(t *testing.T) {
 				Memory: "1Gi",
 			},
 		},
-		EnvironmentName:    "dev",
-		Replicas:           2,
-		TargetNamespace:    namespace,
-		ServiceAccountName: serviceAccountName,
-		DeploymentTimeout:  5 * time.Second,
+		EnvironmentName:          "dev",
+		Replicas:                 2,
+		TargetNamespace:          namespace,
+		ServiceAccountSecretName: serviceAccountSecretName,
+		DeploymentTimeout:        5 * time.Second,
 	}
 	requestResource := corev1.ResourceList{
 		corev1.ResourceCPU:    resource.MustParse("1"),
@@ -317,7 +335,7 @@ func Test_deployer_Deploy(t *testing.T) {
 						Stream:    "stream",
 						Team:      "team",
 					},
-				}, requestResource, limitResource, consumerConfig.ImageName)
+				}, requestResource, limitResource, consumerConfig.ImageName, consumerConfig.ServiceAccountSecretName)
 				prependUpsertDeploymentReactor(t, deploymentAPI, depl, nil, false)
 
 				updatedDepl := changeDeploymentStatus(depl, ready, 1)
@@ -441,7 +459,7 @@ func Test_deployer_Deploy(t *testing.T) {
 						Stream:    "stream",
 						Team:      "team",
 					},
-				}, requestResource, limitResource, consumerConfig.ImageName)
+				}, requestResource, limitResource, consumerConfig.ImageName, consumerConfig.ServiceAccountSecretName)
 				prependUpsertDeploymentReactor(t, deploymentAPI, depl, fmt.Errorf("control plane is down"), false)
 				prependDeleteSecretReactor(t, secretAPI, "project-1-model-1-config", nil)
 				prependDeleteDeploymentReactor(t, deploymentAPI, "project-1-model-1-mlobs", nil)
@@ -526,7 +544,7 @@ func Test_deployer_Deploy(t *testing.T) {
 						Stream:    "stream",
 						Team:      "team",
 					},
-				}, requestResource, limitResource, consumerConfig.ImageName)
+				}, requestResource, limitResource, consumerConfig.ImageName, consumerConfig.ServiceAccountSecretName)
 				preprendGetDeploymentReactor(t, deploymentAPI, depl, nil)
 				prependUpsertDeploymentReactor(t, deploymentAPI, depl, nil, true)
 
@@ -631,7 +649,7 @@ func Test_deployer_Deploy(t *testing.T) {
 						Stream:    "stream",
 						Team:      "team",
 					},
-				}, requestResource, limitResource, consumerConfig.ImageName)
+				}, requestResource, limitResource, consumerConfig.ImageName, consumerConfig.ServiceAccountSecretName)
 				preprendGetDeploymentReactor(t, deploymentAPI, depl, nil)
 				prependUpsertDeploymentReactor(t, deploymentAPI, depl, nil, true)
 
@@ -810,7 +828,7 @@ func Test_deployer_GetDeployedManifest(t *testing.T) {
 		},
 		TopicSource: "caraml-project-1-model-1-1-prediction-log",
 	}
-	depl := createDeploymentSpec(workerData, requestResource, limitResource, "image:v0.1")
+	depl := createDeploymentSpec(workerData, requestResource, limitResource, "image:v0.1", serviceAccountSecretName)
 	testCases := []struct {
 		name             string
 		data             *models.WorkerData

api/pkg/observability/deployment/identity.go

@@ -0,0 +1,55 @@
+package deployment
+
+import (
+	"fmt"
+
+	corev1 "k8s.io/api/core/v1"
+)
+
+func enrichIdentityToPod(podSpec corev1.PodSpec, secretName string, containerNames []string) corev1.PodSpec {
+	secretVolume := createVolumeFromSecret(secretName)
+	updatedPodSpec := podSpec.DeepCopy()
+
+	containerExist := false
+	containerNameLookup := make(map[string]bool)
+	for _, containerName := range containerNames {
+		containerNameLookup[containerName] = true
+	}
+
+	for idx, containerSpec := range updatedPodSpec.Containers {
+		if val := containerNameLookup[containerSpec.Name]; !val {
+			continue
+		}
+
+		containerExist = true
+		mountPath := fmt.Sprintf("/iam/%s", secretName)
+		volumeMount := corev1.VolumeMount{
+			Name:      secretVolume.Name,
+			MountPath: mountPath,
+			ReadOnly:  true,
+		}
+		containerSpec.VolumeMounts = append(containerSpec.VolumeMounts, volumeMount)
+		gcpCredentialEnvVar := corev1.EnvVar{
+			Name:  "GOOGLE_APPLICATION_CREDENTIALS",
+			Value: fmt.Sprintf("%s/service-account.json", mountPath),
+		}
+		containerSpec.Env = append(containerSpec.Env, gcpCredentialEnvVar)
+		updatedPodSpec.Containers[idx] = containerSpec
+	}
+
+	if containerExist {
+		updatedPodSpec.Volumes = append(updatedPodSpec.Volumes, secretVolume)
+	}
+	return *updatedPodSpec
+}
+
+func createVolumeFromSecret(secretName string) corev1.Volume {
+	return corev1.Volume{
+		Name: "iam-secret",
+		VolumeSource: corev1.VolumeSource{
+			Secret: &corev1.SecretVolumeSource{
+				SecretName: secretName,
+			},
+		},
+	}
+}