Use hard-coded AWS user-agent (#146)

spenes · spenes · commit c753a0a2b1ae · 2025-11-21T17:24:12.000+03:00
We don't want to make AWS user-agent configurable anymore.
It will be hard-coded value.
diff --git a/modules/it/src/test/scala/com/snowplowanalytics/snowplow/sources/kinesis/KinesisSinkSpec.scala b/modules/it/src/test/scala/com/snowplowanalytics/snowplow/sources/kinesis/KinesisSinkSpec.scala
@@ -20,7 +20,7 @@ import software.amazon.awssdk.regions.providers.DefaultAwsRegionProviderChain
 import software.amazon.awssdk.regions.Region
 
 import com.snowplowanalytics.snowplow.streams.{ListOfList, Sink, Sinkable}
-import com.snowplowanalytics.snowplow.streams.kinesis.{KinesisFactory, KinesisFactoryConfig}
+import com.snowplowanalytics.snowplow.streams.kinesis.KinesisFactory
 
 import Utils._
 import org.specs2.specification.BeforeAll
@@ -39,7 +39,7 @@ class KinesisSinkSpec extends CatsResource[IO, (Region, LocalStackContainer, Sin
     for {
       region <- Resource.eval(IO.blocking((new DefaultAwsRegionProviderChain).getRegion))
       localstack <- Localstack.resource(region, KINESIS_INITIALIZE_STREAMS, KinesisSinkSpec.getClass.getSimpleName)
-      testFactory <- KinesisFactory.resource[IO](KinesisFactoryConfig(awsUserAgent = None))
+      testFactory <- KinesisFactory.resource[IO]
       testSink <- testFactory.sink(getKinesisSinkConfig(localstack.getEndpoint)(testStream1Name))
     } yield (region, localstack, testSink)
 
diff --git a/modules/it/src/test/scala/com/snowplowanalytics/snowplow/sources/kinesis/KinesisSourceSpec.scala b/modules/it/src/test/scala/com/snowplowanalytics/snowplow/sources/kinesis/KinesisSourceSpec.scala
@@ -22,7 +22,7 @@ import software.amazon.awssdk.regions.providers.DefaultAwsRegionProviderChain
 
 import com.snowplowanalytics.snowplow.streams.EventProcessingConfig
 import com.snowplowanalytics.snowplow.streams.EventProcessingConfig.NoWindowing
-import com.snowplowanalytics.snowplow.streams.kinesis.{KinesisFactory, KinesisFactoryConfig, KinesisSourceConfig}
+import com.snowplowanalytics.snowplow.streams.kinesis.{KinesisFactory, KinesisSourceConfig}
 
 import Utils._
 
@@ -46,7 +46,7 @@ class KinesisSourceSpec
       region <- Resource.eval(IO.blocking((new DefaultAwsRegionProviderChain).getRegion))
       localstack <- Localstack.resource(region, KINESIS_INITIALIZE_STREAMS, KinesisSourceSpec.getClass.getSimpleName)
       kinesisClient <- Resource.eval(getKinesisClient(localstack.getEndpoint, region))
-      kinesisFactory <- KinesisFactory.resource[IO](KinesisFactoryConfig(awsUserAgent = None))
+      kinesisFactory <- KinesisFactory.resource[IO]
     } yield (localstack, kinesisClient, getKinesisSourceConfig(localstack.getEndpoint)(_), kinesisFactory)
 
   override def is = s2"""
diff --git a/modules/kinesis/src/main/resources/reference.conf b/modules/kinesis/src/main/resources/reference.conf
@@ -1,12 +1,4 @@
 snowplow.defaults: {
-  factories: {
-    kinesis: {
-      # Optional AWS user-agent string for tracking cloud spend attribution
-      # No default value for open-source
-      # awsUserAgent: "MyApp/1.0"
-    }
-  }
-
   sources: {
     kinesis: {
       workerIdentifier: ${?HOSTNAME}
diff --git a/modules/kinesis/src/main/scala/com/snowplowanalytics/snowplow/streams/kinesis/KinesisFactory.scala b/modules/kinesis/src/main/scala/com/snowplowanalytics/snowplow/streams/kinesis/KinesisFactory.scala
@@ -16,21 +16,20 @@ import com.snowplowanalytics.snowplow.streams.kinesis.sink.KinesisSink
 import com.snowplowanalytics.snowplow.streams.kinesis.source.KinesisSource
 
 class KinesisFactory[F[_]: Async] private (
-  client: SdkAsyncHttpClient,
-  awsUserAgent: Option[String]
+  client: SdkAsyncHttpClient
 ) extends Factory[F, KinesisSourceConfig, KinesisSinkConfig] {
 
   def sink(config: KinesisSinkConfig): Resource[F, Sink[F]] =
-    KinesisSink.resource(config, client, awsUserAgent)
+    KinesisSink.resource(config, client)
 
   def source(config: KinesisSourceConfig): Resource[F, SourceAndAck[F]] =
-    Resource.eval(KinesisSource.build(config, client, awsUserAgent))
+    Resource.eval(KinesisSource.build(config, client))
 }
 
 object KinesisFactory {
 
-  def resource[F[_]: Async](config: KinesisFactoryConfig): Resource[F, KinesisFactory[F]] =
-    makeClient[F].map(new KinesisFactory(_, config.awsUserAgent))
+  def resource[F[_]: Async]: Resource[F, KinesisFactory[F]] =
+    makeClient[F].map(new KinesisFactory(_))
 
   def makeClient[F[_]: Sync]: Resource[F, SdkAsyncHttpClient] =
     Resource.fromAutoCloseable {
diff --git a/modules/kinesis/src/main/scala/com/snowplowanalytics/snowplow/streams/kinesis/KinesisFactoryConfig.scala b/modules/kinesis/src/main/scala/com/snowplowanalytics/snowplow/streams/kinesis/KinesisFactoryConfig.scala
diff --git a/modules/kinesis/src/main/scala/com/snowplowanalytics/snowplow/streams/kinesis/package.scala b/modules/kinesis/src/main/scala/com/snowplowanalytics/snowplow/streams/kinesis/package.scala
@@ -10,5 +10,7 @@ package com.snowplowanalytics.snowplow.streams
 import cats.Id
 
 package object kinesis {
+  val AWS_USER_AGENT = "APN/1.1 (ak035lu2m8ge2f9qx90duo3ww)"
+
   type KinesisSinkConfig = KinesisSinkConfigM[Id]
 }
diff --git a/modules/kinesis/src/main/scala/com/snowplowanalytics/snowplow/streams/kinesis/sink/KinesisSink.scala b/modules/kinesis/src/main/scala/com/snowplowanalytics/snowplow/streams/kinesis/sink/KinesisSink.scala
@@ -30,7 +30,7 @@ import software.amazon.awssdk.services.kinesis.model.{
 }
 import software.amazon.awssdk.awscore.retry.AwsRetryStrategy
 
-import com.snowplowanalytics.snowplow.streams.kinesis.{BackoffPolicy, KinesisSinkConfig, Retries}
+import com.snowplowanalytics.snowplow.streams.kinesis.{AWS_USER_AGENT, BackoffPolicy, KinesisSinkConfig, Retries}
 
 import java.util.UUID
 import java.nio.charset.StandardCharsets.UTF_8
@@ -43,10 +43,9 @@ private[kinesis] object KinesisSink {
 
   def resource[F[_]: Async](
     config: KinesisSinkConfig,
-    client: SdkAsyncHttpClient,
-    awsUserAgent: Option[String]
+    client: SdkAsyncHttpClient
   ): Resource[F, Sink[F]] =
-    mkProducer[F](config, client, awsUserAgent).map { p =>
+    mkProducer[F](config, client).map { p =>
       new Sink[F] {
         def sink(batch: ListOfList[Sinkable]): F[Unit] =
           writeToKinesis[F](
@@ -66,8 +65,7 @@ private[kinesis] object KinesisSink {
 
   private def mkProducer[F[_]: Sync](
     config: KinesisSinkConfig,
-    client: SdkAsyncHttpClient,
-    awsUserAgent: Option[String]
+    client: SdkAsyncHttpClient
   ): Resource[F, KinesisAsyncClient] =
     Resource.fromAutoCloseable {
       Sync[F].delay {
@@ -81,7 +79,7 @@ private[kinesis] object KinesisSink {
           .defaultsMode(DefaultsMode.AUTO)
           .overrideConfiguration { c =>
             c.retryStrategy(retryStrategy)
-            awsUserAgent.foreach(ua => c.putAdvancedOption(SdkAdvancedClientOption.USER_AGENT_SUFFIX, ua))
+            c.putAdvancedOption(SdkAdvancedClientOption.USER_AGENT_PREFIX, AWS_USER_AGENT)
             ()
           }
         config.customEndpoint.foreach(uri => builder.endpointOverride(uri))
diff --git a/modules/kinesis/src/main/scala/com/snowplowanalytics/snowplow/streams/kinesis/source/KCLScheduler.scala b/modules/kinesis/src/main/scala/com/snowplowanalytics/snowplow/streams/kinesis/source/KCLScheduler.scala
@@ -28,20 +28,19 @@ import java.util.Date
 import java.util.concurrent.{CountDownLatch, LinkedBlockingQueue}
 import java.util.concurrent.atomic.AtomicReference
 
-import com.snowplowanalytics.snowplow.streams.kinesis.KinesisSourceConfig
+import com.snowplowanalytics.snowplow.streams.kinesis.{AWS_USER_AGENT, KinesisSourceConfig}
 
 private[source] object KCLScheduler {
 
   def populateQueue[F[_]: Async](
     config: KinesisSourceConfig,
     queue: LinkedBlockingQueue[KCLAction],
-    client: SdkAsyncHttpClient,
-    awsUserAgent: Option[String]
+    client: SdkAsyncHttpClient
   ): Resource[F, Unit] =
     for {
-      kinesis <- mkKinesisClient[F](config.customEndpoint, client, awsUserAgent)
-      dynamo <- mkDynamoDbClient[F](config.dynamodbCustomEndpoint, client, awsUserAgent)
-      cloudWatch <- mkCloudWatchClient[F](config.cloudwatchCustomEndpoint, client, awsUserAgent)
+      kinesis <- mkKinesisClient[F](config.customEndpoint, client)
+      dynamo <- mkDynamoDbClient[F](config.dynamodbCustomEndpoint, client)
+      cloudWatch <- mkCloudWatchClient[F](config.cloudwatchCustomEndpoint, client)
       scheduler <- Resource.eval(mkScheduler(kinesis, dynamo, cloudWatch, config, queue))
       _ <- runInBackground(scheduler)
     } yield ()
@@ -125,65 +124,56 @@ private[source] object KCLScheduler {
 
   private def mkKinesisClient[F[_]: Sync](
     customEndpoint: Option[URI],
-    client: SdkAsyncHttpClient,
-    awsUserAgent: Option[String]
+    client: SdkAsyncHttpClient
   ): Resource[F, KinesisAsyncClient] =
     Resource.fromAutoCloseable {
       Sync[F].blocking { // Blocking because this might dial the EC2 metadata endpoint
         val builder = KinesisAsyncClient
           .builder()
           .defaultsMode(DefaultsMode.AUTO)
           .httpClient(client)
-        awsUserAgent.foreach { ua =>
-          builder.overrideConfiguration { c =>
-            c.putAdvancedOption(SdkAdvancedClientOption.USER_AGENT_SUFFIX, ua)
+          .overrideConfiguration { c =>
+            c.putAdvancedOption(SdkAdvancedClientOption.USER_AGENT_PREFIX, AWS_USER_AGENT)
             ()
           }
-        }
         customEndpoint.foreach(uri => builder.endpointOverride(uri))
         builder.build()
       }
     }
 
   private def mkDynamoDbClient[F[_]: Sync](
     customEndpoint: Option[URI],
-    client: SdkAsyncHttpClient,
-    awsUserAgent: Option[String]
+    client: SdkAsyncHttpClient
   ): Resource[F, DynamoDbAsyncClient] =
     Resource.fromAutoCloseable {
       Sync[F].blocking { // Blocking because this might dial the EC2 metadata endpoint
         val builder = DynamoDbAsyncClient
           .builder()
           .defaultsMode(DefaultsMode.AUTO)
           .httpClient(client)
-        awsUserAgent.foreach { ua =>
-          builder.overrideConfiguration { c =>
-            c.putAdvancedOption(SdkAdvancedClientOption.USER_AGENT_SUFFIX, ua)
+          .overrideConfiguration { c =>
+            c.putAdvancedOption(SdkAdvancedClientOption.USER_AGENT_PREFIX, AWS_USER_AGENT)
             ()
           }
-        }
         customEndpoint.foreach(uri => builder.endpointOverride(uri))
         builder.build
       }
     }
 
   private def mkCloudWatchClient[F[_]: Sync](
     customEndpoint: Option[URI],
-    client: SdkAsyncHttpClient,
-    awsUserAgent: Option[String]
+    client: SdkAsyncHttpClient
   ): Resource[F, CloudWatchAsyncClient] =
     Resource.fromAutoCloseable {
       Sync[F].blocking { // Blocking because this might dial the EC2 metadata endpoint
         val builder = CloudWatchAsyncClient
           .builder()
           .defaultsMode(DefaultsMode.AUTO)
           .httpClient(client)
-        awsUserAgent.foreach { ua =>
-          builder.overrideConfiguration { c =>
-            c.putAdvancedOption(SdkAdvancedClientOption.USER_AGENT_SUFFIX, ua)
+          .overrideConfiguration { c =>
+            c.putAdvancedOption(SdkAdvancedClientOption.USER_AGENT_PREFIX, AWS_USER_AGENT)
             ()
           }
-        }
         customEndpoint.foreach(uri => builder.endpointOverride(uri))
         builder.build
       }
diff --git a/modules/kinesis/src/main/scala/com/snowplowanalytics/snowplow/streams/kinesis/source/KinesisSource.scala b/modules/kinesis/src/main/scala/com/snowplowanalytics/snowplow/streams/kinesis/source/KinesisSource.scala
@@ -29,15 +29,11 @@ private[kinesis] object KinesisSource {
 
   private implicit def logger[F[_]: Sync]: SelfAwareStructuredLogger[F] = Slf4jLogger.getLogger[F]
 
-  def build[F[_]: Async](
-    config: KinesisSourceConfig,
-    client: SdkAsyncHttpClient,
-    awsUserAgent: Option[String]
-  ): F[SourceAndAck[F]] =
+  def build[F[_]: Async](config: KinesisSourceConfig, client: SdkAsyncHttpClient): F[SourceAndAck[F]] =
     LowLevelSource.toSourceAndAck {
       new LowLevelSource[F, Map[String, Checkpointable]] {
         def stream: Stream[F, Stream[F, Option[LowLevelEvents[Map[String, Checkpointable]]]]] =
-          kinesisStream(config, client, awsUserAgent)
+          kinesisStream(config, client)
 
         def checkpointer: KinesisCheckpointer[F] =
           new KinesisCheckpointer[F](config.checkpointThrottledBackoffPolicy)
@@ -48,12 +44,11 @@ private[kinesis] object KinesisSource {
 
   private def kinesisStream[F[_]: Async](
     config: KinesisSourceConfig,
-    client: SdkAsyncHttpClient,
-    awsUserAgent: Option[String]
+    client: SdkAsyncHttpClient
   ): Stream[F, Stream[F, Option[LowLevelEvents[Map[String, Checkpointable]]]]] = {
     val actionQueue = new LinkedBlockingQueue[KCLAction]()
     for {
-      _ <- Stream.resource(KCLScheduler.populateQueue[F](config, actionQueue, client, awsUserAgent))
+      _ <- Stream.resource(KCLScheduler.populateQueue[F](config, actionQueue, client))
       events <- Stream.emit(pullFromQueueAndEmit(actionQueue).stream).repeat
     } yield events
   }
diff --git a/modules/kinesis/src/test/scala/com/snowplowanalytics/snowplow/streams/kinesis/KinesisFactoryConfigSpec.scala b/modules/kinesis/src/test/scala/com/snowplowanalytics/snowplow/streams/kinesis/KinesisFactoryConfigSpec.scala

Original file line number	Diff line number	Diff line change
`@@ -10,5 +10,7 @@ package com.snowplowanalytics.snowplow.streams`
`10`	`10`	`import cats.Id`
`11`	`11`
`12`	`12`	`package object kinesis {`
	`13`	`+ val AWS_USER_AGENT = "APN/1.1 (ak035lu2m8ge2f9qx90duo3ww)"`
	`14`	`+`
`13`	`15`	`type KinesisSinkConfig = KinesisSinkConfigM[Id]`
`14`	`16`	`}`