Support ObservabilitySettings (#180)

Author: george-zubrienko

.helm/templates/crd-microsoft-synapse.yaml

@@ -71,6 +71,18 @@ spec:
                         - ORC
                         - AVRO
                       default: PARQUET
+                observabilitySettings:
+                  type: object
+                  description: Configurations for enhancing logs and metrics produced by the plugin application
+                  properties:
+                    metricTags:
+                      type: object
+                      description: Additional tags to be assigned to all emitted metrics
+                      additionalProperties:
+                        type: string
+                      default: {}
+                  default:
+                    metricTags: {}
                 sourceSettings:
                   type: object
                   properties:
@@ -84,6 +96,26 @@ spec:
                       description: How long to wait before polling for next change set. Accepted range is between 1s and 3600s
                       minimum: 1
                       maximum: 3600
+                    httpClientMaxRetries:
+                      type: integer
+                      description: Maximum number of retries for the HTTP client
+                      default: 3
+                    httpClientMinRetryDelaySeconds:
+                      type: number
+                      description: Minimum delay between blob client call retries
+                      default: 0.1
+                    httpClientMaxRetryDelaySeconds:
+                      type: number
+                      description: Maximum delay between blob client call retries
+                      default: 1
+                    httpRetryTimeoutSeconds:
+                      type: number
+                      description: Retry timeout for blob storage operations
+                      default: 60
+                    maxResultsPerPage:
+                      type: integer
+                      description: Maximum number of objects per page in paginated responses
+                      default: 5000
                 connectionStringRef:
                   description: |
                     Name of the secret containing connection details to Synapse storage account and Trino.
@@ -113,12 +145,6 @@ spec:
                       type: string
                     apiGroup:
                       type: string
-                httpClientMaxRetries:
-                  type: integer
-                  description: Max number of retries on blob reads for the http client.
-                httpClientRetryDelaySeconds:
-                  type: integer
-                  description: Max retry delay on blob reads for the http client.
                 rowsPerGroup:
                   type: integer
                   description: Maximum number of rows to be grouped together for the staging process to consume.

build.sbt

@@ -25,7 +25,7 @@ lazy val plugin = (project in file("."))
     name := "arcane-stream-microsoft-synapse-link",
     idePackagePrefix := Some("com.sneaksanddata.arcane.microsoft_synapse_link"),
 
-    libraryDependencies += "com.sneaksanddata" % "arcane-framework_3" % "2.0.0-9-gdc66fbd",
+    libraryDependencies += "com.sneaksanddata" % "arcane-framework_3" % "2.1.0",
     libraryDependencies += "io.netty" % "netty-tcnative-boringssl-static" % "2.0.74.Final",
 
     // bugfix for upgrade header

docs/crd.md

@@ -42,6 +42,14 @@ spec:
   # The maximum number of rows in a group
   rowsPerGroup: 10000
 
+  # enhancements for logging and metrics middleware
+  observabilitySettings:
+    # extra metric tags for ALL metrics emitted from the streaming app
+    # Note that tag name and value should only contain alphanumeric characters, separated by `_` if needed 
+    metricTags:
+      key1: value0
+      key2: value1
+
   # The settings of the source table
   sourceSettings:
 
@@ -55,12 +63,21 @@ spec:
     # The root location of the Microsoft Synapse export
     baseLocation: abfss://container@storage-account.dfs.core.windows.net/
 
-  # The maximum number of retries for the HTTP client
-  httpClientMaxRetries: 10
+    # The maximum number of retries for the HTTP client
+    httpClientMaxRetries: 10
+
+    # Minimum delay between blob client call retries
+    httpClientMinRetryDelaySeconds: 0.1
+
+    # Maximum delay between blob client call retries
+    httpClientMaxRetryDelaySeconds: 1
+
+    # Retry timeout for blob storage operations
+    httpRetryTimeoutSeconds: 60
+
+    # Maximum number of objects per page in paginated responses
+    maxResultsPerPage: 5000
 
-  # Delay in seconds between the retries  
-  httpClientRetryDelaySeconds: 5
-   
   # The staging data settings
   stagingDataSettings:
 

src/main/scala/models/app/contracts/StreamSpec.scala

@@ -4,6 +4,12 @@ package models.app.contracts
 import com.sneaksanddata.arcane.framework.models.settings.{TableFormat, TablePropertiesSettings as TableProperties}
 import upickle.default.*
 
+/** Additional logging and metrics configurations
+  */
+case class ObservabilitySettings(
+    metricTags: Map[String, String]
+) derives ReadWriter
+
 /** The configuration of Iceberg catalog
   */
 case class CatalogSettings(
@@ -55,7 +61,16 @@ case class SinkSettings(
 
 /** The configuration of the stream source.
   */
-case class SourceSettings(name: String, baseLocation: String, changeCaptureIntervalSeconds: Int) derives ReadWriter
+case class SourceSettings(
+    name: String,
+    baseLocation: String,
+    changeCaptureIntervalSeconds: Int,
+    httpClientMaxRetries: Int,
+    httpClientMinRetryDelaySeconds: Double,
+    httpClientMaxRetryDelaySeconds: Double,
+    httpRetryTimeoutSeconds: Int,
+    maxResultsPerPage: Int
+) derives ReadWriter
 
 case class TablePropertiesSettings(
     partitionExpressions: Array[String],
@@ -83,6 +98,7 @@ case class FieldSelectionRuleSpec(ruleType: String, fields: Array[String]) deriv
   *   The grouping interval in seconds
   */
 case class StreamSpec(
+    observabilitySettings: ObservabilitySettings,
     sourceSettings: SourceSettings,
 
     // Grouping settings

src/test/scala/integration/StreamRunner.scala

@@ -67,10 +67,21 @@ object StreamRunner extends ZIOSpecDefault:
     |      "catalogUri": "http://localhost:20001/catalog"
     |    }
     |  },
+    |  "observabilitySettings": {
+    |    "metricTags": {
+    |      "key1": "value0",
+    |      "key2": "value1"
+    |    }
+    |  },
     |  "sourceSettings": {
     |    "baseLocation": "abfss://cdm-e2e@devstoreaccount1.dfs.core.windows.net/",
     |    "changeCaptureIntervalSeconds": 300,
-    |    "name": "dimensionattributelevelvalue"
+    |    "name": "dimensionattributelevelvalue",
+    |    "httpClientMaxRetries": 3,
+    |    "httpClientMinRetryDelaySeconds": 0.1,
+    |    "httpClientMaxRetryDelaySeconds": 1,
+    |    "httpRetryTimeoutSeconds": 60,
+    |    "maxResultsPerPage": 5000
     |   },
     |  "stagingDataSettings": {
     |    "catalog": {