Start emitting metrics for creating, expiring and deleting secrets. (…

…#12201)

airbyte-commons-server/src/test/java/io/airbyte/commons/server/handlers/WorkspacesHandlerTest.java

@@ -60,6 +60,7 @@
 import io.airbyte.config.secrets.SecretsRepositoryWriter;
 import io.airbyte.config.secrets.persistence.SecretPersistence;
 import io.airbyte.data.services.WorkspaceService;
+import io.airbyte.metrics.lib.MetricClient;
 import io.airbyte.validation.json.JsonValidationException;
 import java.io.IOException;
 import java.util.Collections;
@@ -122,7 +123,7 @@ void setUp() {
     organizationPersistence = mock(OrganizationPersistence.class);
     secretPersistence = mock(SecretPersistence.class);
     permissionPersistence = mock(PermissionPersistence.class);
-    secretsRepositoryWriter = new SecretsRepositoryWriter(secretPersistence);
+    secretsRepositoryWriter = new SecretsRepositoryWriter(secretPersistence, mock(MetricClient.class));
     connectionsHandler = mock(ConnectionsHandler.class);
     destinationHandler = mock(DestinationHandler.class);
     sourceHandler = mock(SourceHandler.class);

airbyte-config/config-secrets/build.gradle.kts

@@ -34,6 +34,7 @@ dependencies {
    */
   implementation(project(":airbyte-config:config-models"))
   implementation(project(":airbyte-json-validation"))
+  implementation(project(":airbyte-metrics:metrics-lib"))
 
   testAnnotationProcessor(platform(libs.micronaut.platform))
   testAnnotationProcessor(libs.bundles.micronaut.test.annotation.processor)
@@ -55,4 +56,4 @@ afterEvaluate {
   tasks.named("kaptGenerateStubsTestKotlin") {
     enabled = false
   }
-}
+}

airbyte-config/config-secrets/src/main/kotlin/secrets/SecretsRepositoryWriter.kt

@@ -7,6 +7,8 @@ package io.airbyte.config.secrets
 import com.fasterxml.jackson.databind.JsonNode
 import io.airbyte.config.secrets.persistence.RuntimeSecretPersistence
 import io.airbyte.config.secrets.persistence.SecretPersistence
+import io.airbyte.metrics.lib.MetricClient
+import io.airbyte.metrics.lib.OssMetricsRegistry
 import io.airbyte.protocol.models.ConnectorSpecification
 import io.airbyte.validation.json.JsonSchemaValidator
 import io.airbyte.validation.json.JsonValidationException
@@ -32,6 +34,7 @@ private val EPHEMERAL_SECRET_LIFE_DURATION = Duration.ofHours(2)
 @Requires(bean = SecretPersistence::class)
 open class SecretsRepositoryWriter(
   private val secretPersistence: SecretPersistence,
+  private val metricClient: MetricClient,
 ) {
   val validator: JsonSchemaValidator = JsonSchemaValidator()
 
@@ -104,8 +107,9 @@ open class SecretsRepositoryWriter(
     if (validate) {
       validator.ensure(spec, fullConfig)
     }
+    val update = oldConfig.isPresent
     val splitSecretConfig: SplitSecretConfig =
-      if (oldConfig.isPresent) {
+      if (update) {
         SecretsHelpers.splitAndUpdateConfig(
           workspaceId,
           oldConfig.get(),
@@ -121,8 +125,12 @@ open class SecretsRepositoryWriter(
           secretPersistence,
         )
       }
+
     splitSecretConfig.getCoordinateToPayload()
       .forEach { (coordinate: SecretCoordinate, payload: String) ->
+        if (update) {
+          metricClient.count(OssMetricsRegistry.UPDATE_SECRET_DEFAULT_STORE, 1)
+        }
         secretPersistence.write(coordinate, payload)
       }
     return splitSecretConfig.partialConfig

airbyte-config/config-secrets/src/main/kotlin/secrets/persistence/GoogleSecretManagerPersistence.kt

@@ -18,6 +18,10 @@ import com.google.cloud.secretmanager.v1.SecretPayload
 import com.google.cloud.secretmanager.v1.SecretVersionName
 import com.google.protobuf.ByteString
 import io.airbyte.config.secrets.SecretCoordinate
+import io.airbyte.metrics.lib.MetricAttribute
+import io.airbyte.metrics.lib.MetricClient
+import io.airbyte.metrics.lib.MetricTags
+import io.airbyte.metrics.lib.OssMetricsRegistry
 import io.github.oshai.kotlinlogging.KotlinLogging
 import io.micronaut.context.annotation.Requires
 import io.micronaut.context.annotation.Value
@@ -45,6 +49,7 @@ private val logger = KotlinLogging.logger {}
 class GoogleSecretManagerPersistence(
   @Value("\${airbyte.secret.store.gcp.project-id}") val gcpProjectId: String,
   private val googleSecretManagerServiceClient: GoogleSecretManagerServiceClient,
+  private val metricClient: MetricClient,
 ) : SecretPersistence {
   override fun read(coordinate: SecretCoordinate): String {
     try {
@@ -92,11 +97,14 @@ class GoogleSecretManagerPersistence(
       if (read(coordinate).isEmpty()) {
         val secretBuilder = Secret.newBuilder().setReplication(replicationPolicy)
 
+        var expTag = listOf(MetricAttribute(MetricTags.EXPIRE_SECRET, "false"))
         expiry?.let {
           val expireTime = com.google.protobuf.Timestamp.newBuilder().setSeconds(it.epochSecond).build()
           secretBuilder.setExpireTime(expireTime)
+          expTag = listOf(MetricAttribute(MetricTags.EXPIRE_SECRET, "true"))
         }
 
+        metricClient.count(OssMetricsRegistry.CREATE_SECRET_DEFAULT_STORE, 1, *expTag.toTypedArray())
         client.createSecret(ProjectName.of(gcpProjectId), coordinate.fullCoordinate, secretBuilder.build())
       }
 
@@ -110,6 +118,7 @@ class GoogleSecretManagerPersistence(
     googleSecretManagerServiceClient.createClient().use { client ->
       val secretName = SecretName.of(gcpProjectId, coordinate.fullCoordinate)
       client.deleteSecret(secretName)
+      metricClient.count(OssMetricsRegistry.DELETE_SECRET_DEFAULT_STORE, 1)
     }
   }
 }

airbyte-config/config-secrets/src/main/kotlin/secrets/persistence/RuntimeSecretPersistence.kt

@@ -8,6 +8,8 @@ import io.airbyte.config.AwsAccessKeySecretPersistenceConfig
 import io.airbyte.config.AwsRoleSecretPersistenceConfig
 import io.airbyte.config.SecretPersistenceConfig
 import io.airbyte.config.secrets.SecretCoordinate
+import io.airbyte.metrics.lib.MetricClientFactory
+import io.airbyte.metrics.lib.MetricEmittingApps
 import io.github.oshai.kotlinlogging.KotlinLogging
 import io.micronaut.context.annotation.Property
 import kotlin.jvm.optionals.getOrElse
@@ -34,11 +36,13 @@ class RuntimeSecretPersistence(private val secretPersistenceConfig: SecretPersis
       }
 
       SecretPersistenceConfig.SecretPersistenceType.GOOGLE -> {
+        // We cannot use the @Singleton here because this class is not managed by Micronaut.
+        // Manually create the client for now.
+        MetricClientFactory.initialize(MetricEmittingApps.SERVER)
         GoogleSecretManagerPersistence(
           secretPersistenceConfig.configuration["gcpProjectId"]!!,
-          GoogleSecretManagerServiceClient(
-            secretPersistenceConfig.configuration["gcpCredentialsJson"]!!,
-          ),
+          GoogleSecretManagerServiceClient(secretPersistenceConfig.configuration["gcpCredentialsJson"]!!),
+          MetricClientFactory.getMetricClient(),
         )
       }
 

airbyte-config/config-secrets/src/test/kotlin/secrets/SecretsRepositoryWriterTest.kt

@@ -10,6 +10,7 @@ import io.airbyte.config.DestinationConnection
 import io.airbyte.config.SourceConnection
 import io.airbyte.config.persistence.ConfigRepository
 import io.airbyte.config.secrets.hydration.RealSecretsHydrator
+import io.airbyte.metrics.lib.MetricClient
 import io.airbyte.protocol.models.ConnectorSpecification
 import io.airbyte.validation.json.JsonSchemaValidator
 import io.mockk.mockk
@@ -28,9 +29,11 @@ internal class SecretsRepositoryWriterTest {
   fun setup() {
     configRepository = mockk()
     secretPersistence = MemorySecretPersistence()
+    val metricClient: MetricClient = mockk()
     secretsRepositoryWriter =
       SecretsRepositoryWriter(
         secretPersistence,
+        metricClient,
       )
     secretsHydrator = RealSecretsHydrator(secretPersistence)
     secretsRepositoryReader = SecretsRepositoryReader(secretsHydrator)

airbyte-config/config-secrets/src/test/kotlin/secrets/persistence/GoogleSecretManagerPersistenceTest.kt

@@ -17,6 +17,7 @@ import com.google.cloud.secretmanager.v1.SecretVersionName
 import com.google.protobuf.ByteString
 import io.airbyte.config.secrets.SecretCoordinate
 import io.airbyte.config.secrets.persistence.GoogleSecretManagerPersistence.Companion.replicationPolicy
+import io.airbyte.metrics.lib.MetricClient
 import io.grpc.Status
 import io.mockk.Runs
 import io.mockk.every
@@ -38,13 +39,15 @@ class GoogleSecretManagerPersistenceTest {
     val mockGoogleClient: SecretManagerServiceClient = mockk()
     val mockResponse: AccessSecretVersionResponse = mockk()
     val mockPayload: SecretPayload = mockk()
-    val persistence = GoogleSecretManagerPersistence(projectId, mockClient)
+    val mockMetric: MetricClient = mockk()
+    val persistence = GoogleSecretManagerPersistence(projectId, mockClient, mockMetric)
 
     every { mockPayload.data } returns ByteString.copyFromUtf8(secret)
     every { mockResponse.payload } returns mockPayload
     every { mockGoogleClient.accessSecretVersion(ofType(SecretVersionName::class)) } returns mockResponse
     every { mockGoogleClient.close() } returns Unit
     every { mockClient.createClient() } returns mockGoogleClient
+    every { mockMetric.count(any(), any()) } returns Unit
 
     val result = persistence.read(coordinate)
     Assertions.assertEquals(secret, result)
@@ -56,7 +59,8 @@ class GoogleSecretManagerPersistenceTest {
     val coordinate = SecretCoordinate.fromFullCoordinate("secret_coordinate_v1")
     val mockClient: GoogleSecretManagerServiceClient = mockk()
     val mockGoogleClient: SecretManagerServiceClient = mockk()
-    val persistence = GoogleSecretManagerPersistence(projectId, mockClient)
+    val mockMetric: MetricClient = mockk()
+    val persistence = GoogleSecretManagerPersistence(projectId, mockClient, mockMetric)
 
     every { mockGoogleClient.accessSecretVersion(ofType(SecretVersionName::class)) } throws
       NotFoundException(
@@ -68,6 +72,7 @@ class GoogleSecretManagerPersistenceTest {
       )
     every { mockGoogleClient.close() } returns Unit
     every { mockClient.createClient() } returns mockGoogleClient
+    every { mockMetric.count(any(), any()) } returns Unit
 
     Assertions.assertDoesNotThrow {
       val result = persistence.read(coordinate)
@@ -84,7 +89,8 @@ class GoogleSecretManagerPersistenceTest {
     val mockGoogleClient: SecretManagerServiceClient = mockk()
     val mockResponse: AccessSecretVersionResponse = mockk()
     val mockPayload: SecretPayload = mockk()
-    val persistence = GoogleSecretManagerPersistence(projectId, mockClient)
+    val mockMetric: MetricClient = mockk()
+    val persistence = GoogleSecretManagerPersistence(projectId, mockClient, mockMetric)
 
     every { mockPayload.data } returns ByteString.copyFromUtf8(secret)
     every { mockResponse.payload } returns mockPayload
@@ -100,6 +106,7 @@ class GoogleSecretManagerPersistenceTest {
     every { mockGoogleClient.addSecretVersion(any<SecretName>(), any<SecretPayload>()) } returns mockk<SecretVersion>()
     every { mockGoogleClient.close() } returns Unit
     every { mockClient.createClient() } returns mockGoogleClient
+    every { mockMetric.count(any(), any(), any()) } returns Unit
 
     persistence.write(coordinate, secret)
 
@@ -116,7 +123,8 @@ class GoogleSecretManagerPersistenceTest {
     val mockGoogleClient: SecretManagerServiceClient = mockk()
     val mockResponse: AccessSecretVersionResponse = mockk()
     val mockPayload: SecretPayload = mockk()
-    val persistence = GoogleSecretManagerPersistence(projectId, mockClient)
+    val mockMetric: MetricClient = mockk()
+    val persistence = GoogleSecretManagerPersistence(projectId, mockClient, mockMetric)
 
     every { mockPayload.data } returns ByteString.copyFromUtf8(secret)
     every { mockResponse.payload } returns mockPayload
@@ -132,6 +140,7 @@ class GoogleSecretManagerPersistenceTest {
     every { mockGoogleClient.addSecretVersion(any<SecretName>(), any<SecretPayload>()) } returns mockk<SecretVersion>()
     every { mockGoogleClient.close() } returns Unit
     every { mockClient.createClient() } returns mockGoogleClient
+    every { mockMetric.count(any(), any(), any()) } returns Unit
 
     val expiry = Instant.now().plus(Duration.ofMinutes(1))
     persistence.writeWithExpiry(coordinate, secret, expiry)
@@ -153,14 +162,16 @@ class GoogleSecretManagerPersistenceTest {
     val mockGoogleClient: SecretManagerServiceClient = mockk()
     val mockResponse: AccessSecretVersionResponse = mockk()
     val mockPayload: SecretPayload = mockk()
-    val persistence = GoogleSecretManagerPersistence(projectId, mockClient)
+    val mockMetric: MetricClient = mockk()
+    val persistence = GoogleSecretManagerPersistence(projectId, mockClient, mockMetric)
 
     every { mockPayload.data } returns ByteString.copyFromUtf8(secret)
     every { mockResponse.payload } returns mockPayload
     every { mockGoogleClient.accessSecretVersion(ofType(SecretVersionName::class)) } returns mockResponse
     every { mockGoogleClient.addSecretVersion(any<SecretName>(), any<SecretPayload>()) } returns mockk<SecretVersion>()
     every { mockGoogleClient.close() } returns Unit
     every { mockClient.createClient() } returns mockGoogleClient
+    every { mockMetric.count(any(), any(), any()) } returns Unit
 
     persistence.write(coordinate, secret)
 
@@ -176,13 +187,15 @@ class GoogleSecretManagerPersistenceTest {
     val mockGoogleClient: SecretManagerServiceClient = mockk()
     val mockResponse: AccessSecretVersionResponse = mockk()
     val mockPayload: SecretPayload = mockk()
-    val persistence = GoogleSecretManagerPersistence(projectId, mockClient)
+    val mockMetric: MetricClient = mockk()
+    val persistence = GoogleSecretManagerPersistence(projectId, mockClient, mockMetric)
 
     every { mockPayload.data } returns ByteString.copyFromUtf8(secret)
     every { mockResponse.payload } returns mockPayload
     every { mockClient.createClient() } returns mockGoogleClient
     every { mockGoogleClient.deleteSecret(ofType(SecretName::class)) } just Runs
     every { mockGoogleClient.close() } returns Unit
+    every { mockMetric.count(any(), any()) } returns Unit
 
     persistence.delete(coordinate)
 

airbyte-metrics/metrics-lib/src/main/java/io/airbyte/metrics/lib/MetricTags.java

@@ -27,6 +27,8 @@ public class MetricTags {
   public static final String CRON_TYPE = "cron_type";
   public static final String DESTINATION_ID = "destination_id";
   public static final String DESTINATION_IMAGE = "destination_image";
+  public static final String EXPIRE_SECRET = "expire_secret";
+  public static final String FAILURE_CAUSE = "failure_cause";
   public static final String FAILURE_ORIGIN = "failure_origin";
   public static final String FAILURE_TYPE = "failure_type";
   public static final String GEOGRAPHY = "geography";
@@ -46,7 +48,6 @@ public class MetricTags {
   public static final String NOTIFICATION_CLIENT = "notification_client";
   public static final String RECORD_COUNT_TYPE = "record_count_type";
   public static final String RELEASE_STAGE = "release_stage";
-  public static final String FAILURE_CAUSE = "failure_cause";
   public static final String SOURCE_ID = "source_id";
   public static final String SOURCE_IMAGE = "source_image";
   public static final String STATUS = "status";

airbyte-metrics/metrics-lib/src/main/java/io/airbyte/metrics/lib/OssMetricsRegistry.java

@@ -450,7 +450,17 @@ public enum OssMetricsRegistry implements MetricsRegistry {
 
   PAYLOAD_VALIDATION_RESULT(MetricEmittingApps.WORKER,
       "payload_validation_result",
-      "The result of the comparing the payload in object storage to the one passed from temporal.");
+      "The result of the comparing the payload in object storage to the one passed from temporal."),
+
+  CREATE_SECRET_DEFAULT_STORE(MetricEmittingApps.SERVER,
+      "create_secret_default_store",
+      "A secret was created in the default configured secret store."),
+  UPDATE_SECRET_DEFAULT_STORE(MetricEmittingApps.SERVER,
+      "create_secret_default_store",
+      "A secret was created in the default configured secret store."),
+  DELETE_SECRET_DEFAULT_STORE(MetricEmittingApps.SERVER,
+      "delete_secret_default_store",
+      "A secret was created in the default configured secret store.");
 
   private final MetricEmittingApp application;
   private final String metricName;