Pubsub sink #327
Pubsub sink #327
Conversation
pondzix
force-pushed
the
http4s/pubsub_sink
branch
from
c71eb38
to
e7251c9
Compare
benjben
force-pushed
the
http4s_config
branch
4 times, most recently
from
dbbd023
to
76a3474
Compare
pondzix
force-pushed
the
http4s/pubsub_sink
branch
2 times, most recently
from
f577884
to
9deaf81
Compare
| } | ||
| } | ||
|
|
||
| private def parse(resource: String): Either[ExitCode, Config[PubSubSinkConfig]] = { |
There was a problem hiding this comment.
Could be put in http4s module ?
There was a problem hiding this comment.
I think I would keep it here for now. When we add more sinks and we indeed keep repeating this piece of code, we can extract to common shared module.
|
|
||
| private def parse(resource: String): Either[ExitCode, Config[PubSubSinkConfig]] = { | ||
| val path = Paths.get(getClass.getResource(resource).toURI) | ||
| ConfigParser.fromPath[IO, PubSubSinkConfig](Some(path)).value.unsafeRunSync() |
There was a problem hiding this comment.
Could we use CatsEffect to avoid this unsafeRunSync() ?
| object BuilderOps { | ||
|
|
||
| implicit class PublisherBuilderOps(val builder: Publisher.Builder) extends AnyVal { | ||
| def setProvidersForEmulator(): Publisher.Builder = | ||
| customEmulatorHost().fold(builder) { emulatorHost => | ||
| builder | ||
| .setChannelProvider(createCustomChannelProvider(emulatorHost)) | ||
| .setCredentialsProvider(NoCredentialsProvider.create()) | ||
| } | ||
| } | ||
|
|
||
| implicit class TopicAdminBuilderOps(val builder: TopicAdminSettings.Builder) extends AnyVal { | ||
| def setProvidersForEmulator(): TopicAdminSettings.Builder = | ||
| customEmulatorHost().fold(builder) { emulatorHost => | ||
| builder | ||
| .setTransportChannelProvider(createCustomChannelProvider(emulatorHost)) | ||
| .setCredentialsProvider(NoCredentialsProvider.create()) | ||
| } | ||
| } | ||
|
|
||
| private def customEmulatorHost(): Option[String] = | ||
| sys.env.get("PUBSUB_EMULATOR_HOST") | ||
|
|
||
| private def createCustomChannelProvider(emulatorHost: String): FixedTransportChannelProvider = { | ||
| val channel = ManagedChannelBuilder.forTarget(emulatorHost).usePlaintext().build() | ||
| FixedTransportChannelProvider.create(GrpcTransportChannel.create(channel)) | ||
| } | ||
| } |
| } | ||
|
|
||
| object PubSubSink { | ||
| private val UserAgent = s"snowplow/stream-collector-${BuildInfo.version}" |
There was a problem hiding this comment.
| private val UserAgent = s"snowplow/stream-collector-${BuildInfo.version}" | |
| private val UserAgent = s"${BuildInfo.dockerAlias}:${BuildInfo.version}" |
There was a problem hiding this comment.
Shouldn't that be just BuildInfo.dockerAlias? It looks like it already contains version
| @@ -67,7 +67,7 @@ object ConfigParser { | |||
| } | |||
|
|
|||
| private def loadAll(config: TypesafeConfig): TypesafeConfig = | |||
| namespaced(ConfigFactory.load(namespaced(config.withFallback(namespaced(ConfigFactory.load()))))) | |||
| namespaced(config.withFallback(namespaced(ConfigFactory.load()))) | |||
| healthCheckTask: F[Unit] | ||
| ): Resource[F, Unit] = { | ||
| val checkThenSleep = healthCheckTask *> Async[F].sleep(sinkConfig.startupCheckInterval) | ||
| checkThenSleep.untilM_(isHealthyState.get).background.void |
.../src/main/scala/com.snowplowanalytics.snowplow.collectors.scalastream/sinks/PubSubSink.scala
Outdated
Show resolved
Hide resolved
| ) extends Sink[F] { | ||
|
|
||
| override def storeRawEvents(events: List[Array[Byte]], key: String): F[Unit] = | ||
| produceBatch(events).start.void |
There was a problem hiding this comment.
If the tracker doesn't batch the events, events will always be of size 1. But nevertheless this seems fine to me to start a Fiber for each single event.
| ) | ||
|
|
||
| private def handlePublishError(error: Throwable): F[Unit] = | ||
| isHealthyState.set(false) *> Logger[F].error(createErrorMessage(error)) |
There was a problem hiding this comment.
Related to the discussion about logging policy, we'll need to decide if we should be logging errors for individual events or if we need some batching.
There was a problem hiding this comment.
Good point. This only becomes an issue if there is an outage and if it lasts for a long time. Ideally we should find a nice pattern in the code for de-duping log messages upon errors. And we should start using it everywhere when we've decided on that pattern.
No need to hold up this current PR though.
pondzix
force-pushed
the
http4s/pubsub_sink
branch
6 times, most recently
from
f5f9e72
to
656ccd6
Compare
pondzix
force-pushed
the
http4s/pubsub_sink
branch
from
656ccd6
to
cfdaa58
Compare
|
|
||
| private def createTopicAdminClient[F[_]: Sync](): Resource[F, TopicAdminClient] = { | ||
| val builder = TopicAdminSettings.newBuilder().setProvidersForEmulator().build() | ||
| Resource.make(Sync[F].delay(TopicAdminClient.create(builder)))(client => Sync[F].delay(client.close())) |
There was a problem hiding this comment.
Do you know if client.close() is blocking or non-blocking?
| ) | ||
|
|
||
| private def handlePublishError(error: Throwable): F[Unit] = | ||
| isHealthyState.set(false) *> Logger[F].error(createErrorMessage(error)) |
There was a problem hiding this comment.
Good point. This only becomes an issue if there is an outage and if it lasts for a long time. Ideally we should find a nice pattern in the code for de-duping log messages upon errors. And we should start using it everywhere when we've decided on that pattern.
No need to hold up this current PR though.
corepart of integration tests based on CE2 has been copied to the newhttp4smodule and ported to CE3.