Configuration

License

Enrich is released under the Snowplow Limited Use License (FAQ).

To accept the terms of license and run Enrich, set the ACCEPT_LIMITED_USE_LICENSE=yes environment variable. Alternatively, you can configure the license.accept option, like this:

license {
  accept = true
}

Common parameters

parameter	description
concurrency.enrich	Optional. Default: 256. Number of events that can get enriched at the same time within a chunk (events are processed by chunks in the app).
concurrency.sink	Optional. Default for enrich-pubsub: 3. Default for enrich-kinesis: 1. Number of chunks that can get sunk at the same time. WARNING for enrich-kinesis: if greater than 1, records can get checkpointed before they are sunk.
assetsUpdatePeriod	Optional. E.g. 7 days. Period after which enrich assets (e.g. the maxmind database for the IpLookups enrichment) should be checked for udpates. Assets will never be updated if this key is missing.
monitoring.sentry.dsn	Optional. E.g. http://sentry.acme.com. To track uncaught runtime exceptions in Sentry.
monitoring.metrics.statsd.hostname	Optional. E.g. localhost. Hostname of the StatsD server to send enrichment metrics (latency and event counts) to.
monitoring.metrics.statsd.port	Optional. E.g. 8125. Port of the StatsD server.
monitoring.metrics.statsd.period	Optional. E.g. 10 seconds. How frequently to send metrics to StatsD server.
monitoring.metrics.statsd.tags	Optional. E.g. { "env": "prod" }. Key-value pairs attached to each metric sent to StatsD to provide contextual information.
monitoring.metrics.statsd.prefix	Optional. Default: snowplow.enrich. Pefix of StatsD metric names.
monitoring.metrics.stdout.period	Optional. E.g. 10 seconds. If set, metrics will be printed in the logs with this frequency.
monitoring.metrics.stdout.prefix	Optional. Default: snowplow.enrich. Prefix for the metrics appearing in the logs.
telemetry.disable	Optional. Set to true to disable telemetry.
telemetry.userProvidedId	Optional. See here for more information.
featureFlags.acceptInvalid	Optional. Default: false. Enrich 3.0.0 introduces the validation of the enriched events against atomic schema before emitting. If set to false, a failed event will be emitted instead of the enriched event if validation fails. If set to true, invalid enriched events will be emitted, as before.
validation.atomicFieldsLimits (since 4.0.0)	Optional. For the defaults, see here. Configuration for custom maximum atomic fields (strings) length. It's a map-like structure with keys being atomic field names and values being their max allowed length.

Instead of a message queue, it's also possible to read collector payloads from files on disk. This can be used for instance for testing purposes. In this case the configuration needs to be as below.

parameter	description
input.type	Required. Must be FileSystem.
input.dir	Required. E.g. /input/collectorPayloads/. Directory containing collector payloads encoded with Thrift.

Likewise, it's possible to write enriched events, pii events and failed events to files instead of PubSub or Kinesis.

To write enriched events to files:

parameter	description
output.good.type	Required. Must be FileSystem.
output.good.file	Required. E.g. /output/enriched. File where enriched events will be written.
output.good.maxBytes	Optional. E.g. 1048576. Maximum size of a file in bytes. Triggers file rotation.

To write failed events to files:

parameter	description
output.bad.type	Required. Must be FileSystem.
output.bad.file	Required. E.g. /output/bad. File where failed events will be written.
output.bad.maxBytes	Optional. E.g. 1048576. Maximum size of a file in bytes. Triggers file rotation.

To write pii events to files:

parameter	description
output.pii.type	Required. Must be FileSystem.
output.pii.file	Required. E.g. /output/pii. File where pii events will be written.
output.pii.maxBytes	Optional. E.g. 1048576. Maximum size of a file in bytes. Triggers file rotation.

enrich-pubsub

A minimal configuration file can be found on the Github repo, as well as a comprehensive one.

parameter	description
input.subscription	Required. E.g. projects/example-project/subscriptions/collectorPayloads. PubSub subscription identifier for the collector payloads.
input.parallelPullCount	Optional. Default: 1. Number of threads used internally by permutive library to handle incoming messages. These threads do very little "work" apart from writing the message to a concurrent queue.
input.maxQueueSize	Optional. Default: 3000. Configures the "max outstanding element count" of PubSub. This is the principal way we control concurrency in the app; it puts an upper bound on the number of events in memory at once. An event counts towards this limit starting from when it received by the permutive library, until we ack it (after publishing to output). The value must be large enough that it does not cause the sink to block whilst it is waiting for a batch to be completed. The first of maxQueueSize and maxRequestBytes being reached will pause the consumption.
input.maxRequestBytes	Optional. Default: 50000000 (50MB). Configures the "maximum outstanding request bytes" of PubSub subscriber. It puts an upper bound on the events' bytes that can be hold in memory at once before getting acked. The value must be large enough to not cause the sink to block whilst it is waiting for a batch to be completed. The first of maxQueueSize and maxRequestBytes being reached will pause the consumption.
input.maxAckExtensionPeriod	Optional. Default: 1 hour. Maximum period a message ack deadline can be extended. A zero duration disables auto deadline extension.
output.good.topic	Required. E.g. projects/example-project/topics/enriched. Name of the PubSub topic that will receive the enriched events.
output.good.attributes	Optional. Enriched event fields to add as PubSub message attributes. For example, if this is [ "app_id" ] then the enriched event's app_id field will be an attribute of the PubSub message, as well as being a field within the enriched event.
output.good.delayThreshold	Optional. Default: 200 milliseconds. Delay threshold to use for batching. After this amount of time has elapsed, before maxBatchSize and maxBatchBytes have been reached, messages from the buffer will be sent.
output.good.maxBatchSize	Optional. Default: 1000 (PubSub maximum). Maximum number of messages sent within a batch. When the buffer reaches this number of messages they are sent.
output.good.maxBatchBytes	Optional. Default: 8000000 (PubSub maximum is 10MB). Maximum number of bytes sent within a batch. When the buffer reaches this size messages are sent.
output.incomplete.topic	Required. E.g. projects/example-project/topics/incomplete. Name of the PubSub topic that will receive the failed events (same format as the enriched events).
output.incomplete.delayThreshold	Same as output.good.delayThreshold for failed events.
output.incomplete.maxBatchSize	Same as output.good.maxBatchSize for failed events.
output.incomplete.maxBatchBytes	Same as output.good.maxBatchBytes for failed events.
output.bad.topic	Required. E.g. projects/example-project/topics/bad. Name of the PubSub topic that will receive the failed events in the "bad row" format (JSON).
output.bad.delayThreshold	Same as output.good.delayThreshold for failed events in the "bad row" format (JSON).
output.bad.maxBatchSize	Same as output.good.maxBatchSize for failed events in the "bad row" format (JSON).
output.bad.maxBatchBytes	Same as output.good.maxBatchBytes for failed events in the "bad row" format (JSON).
output.pii.topic	Optional. Example: projects/test-project/topics/pii. Should be used in conjunction with the PII pseudonymization enrichment. When configured, enables an extra output topic for writing a pii_transformation event.
output.pii.attributes	Same as output.good.attributes for pii events.
output.pii.delayThreshold	Same as output.good.delayThreshold for pii events.
output.pii.maxBatchSize	Same as output.good.maxBatchSize for pii events.
output.pii.maxBatchBytes	Same as output.good.maxBatchBytes for pii events.

enrich-kinesis

A minimal configuration file can be found on the Github repo, as well as a comprehensive one.

parameter	description
input.appName	Optional. Default: snowplow-enrich-kinesis. Name of the application which the KCL daemon should assume. A DynamoDB table with this name will be created.
input.streamName	Required. E.g. raw. Name of the Kinesis stream with the collector payloads to read from.
input.region	Optional. E.g. eu-central-1. Region where the Kinesis stream is located. This field is optional if it can be resolved with AWS region provider chain. It checks places like env variables, system properties, AWS profile file.
input.initialPosition.type	Optional. Default: TRIM_HORIZON. Set the initial position to consume the Kinesis stream. Possible values: LATEST (most recent data), TRIM_HORIZON (oldest available data), AT_TIMESTAMP (start from the record at or after the specified timestamp).
input.initialPosition.timestamp	Required for AT_TIMESTAMP. E.g. 2020-07-17T10:00:00Z.
input.retrievalMode.type	Optional. Default: Polling. Set the mode for retrieving records. Possible values: Polling or FanOut.
input.retrievalMode.maxRecords	Required for Polling. Default: 10000. Maximum size of a batch returned by a call to getRecords. Records are checkpointed after a batch has been fully processed, thus the smaller maxRecords, the more often records can be checkpointed into DynamoDb, but possibly reducing the throughput.
input.bufferSize	Optional. Default: 3. Size of the internal buffer used when reading messages from Kinesis, each buffer holding up to maxRecords from above.
input.customEndpoint	Optional. E.g. http://localhost:4566. Endpoint url configuration to override aws kinesis endpoints. Can be used to specify local endpoint when using localstack.
input.dynamodbCustomEndpoint	Optional. E.g. http://localhost:4566. Endpoint url configuration to override aws dyanomdb endpoint for Kinesis checkpoints lease table. Can be used to specify local endpoint when using localstack.
input.cloudwatchCustomEndpoint	Optional. E.g. http://localhost:4566. Endpoint url configuration to override aws cloudwatch endpoint for metrics. Can be used to specify local endpoint when using localstack.
output.good.streamName	Required. E.g. enriched. Name of the Kinesis stream to write to the enriched events.
output.good.region	Same as input.region for enriched events stream.
output.good.partitionKey	Optional. How the output stream will be partitioned in Kinesis. Events with the same partition key value will go to the same shard. Possible values: event_id, event_fingerprint, domain_userid, network_userid, user_ipaddress, domain_sessionid, user_fingerprint. If not specified, the partition key will be a random UUID.
output.good.backoffPolicy.minBackoff	Optional. Default: 100 milliseconds. Minimum backoff before retrying when writing fails with internal errors.
output.good.backoffPolicy.maxBackoff	Optional. Default: 10 seconds. Maximum backoff before retrying when writing fails with internal errors.
output.good.backoffPolicy.maxRetries	Optional. Default: 10. Maximum number of retries for internal errors.
output.good.throttledBackoffPolicy.minBackoff (since 3.4.1)	Optional. Default: 100 milliseconds. Minimum backoff before retrying when writing fails in case of throughput exceeded.
output.good.throttledBackoffPolicy.maxBackoff (since 3.4.1)	Optional. Default: 1 second. Maximum backoff before retrying when writing fails in case of throughput exceeded. Writing is retried forever.
output.good.recordLimit	Optional. Default: 500 (maximum allowed). Limits the number of events in a single PutRecords request. Several requests are made in parallel.
output.good.customEndpoint	Optional. E.g. http://localhost:4566. To use a custom Kinesis endpoint.
output.incomplete.streamName	Required. E.g. incomplete. Name of the Kinesis stream that will receive the failed events (same format as the enriched events).
output.incomplete.region	Same as output.good.region for failed events.
output.incomplete.backoffPolicy.minBackoff	Same as output.good.backoffPolicy.minBackoff for failed events.
output.incomplete.backoffPolicy.maxBackoff	Same as output.good.backoffPolicy.maxBackoff for failed events.
output.incomplete.backoffPolicy.maxRetries	Same as output.good.backoffPolicy.maxRetries for failed events.
output.incomplete.throttledBackoffPolicy.minBackoff (since 3.4.1)	Same as output.good.throttledBackoffPolicy.minBackoff for failed events.
output.incomplete.throttledBackoffPolicy.maxBackoff (since 3.4.1)	Same as output.good.throttledBackoffPolicy.maxBackoff for failed events.
output.incomplete.recordLimit	Same as output.good.recordLimit for failed events.
output.incomplete.customEndpoint	Same as output.good.customEndpoint for failed events.
output.bad.streamName	Required. E.g. bad. Name of the Kinesis stream that will receive the failed events in the "bad row" format (JSON).
output.bad.region	Same as output.good.region for failed events in the "bad row" format (JSON).
output.bad.backoffPolicy.minBackoff	Same as output.good.backoffPolicy.minBackoff for failed events in the "bad row" format (JSON).
output.bad.backoffPolicy.maxBackoff	Same as output.good.backoffPolicy.maxBackoff for failed events in the "bad row" format (JSON).
output.bad.backoffPolicy.maxRetries	Same as output.good.backoffPolicy.maxRetries for failed events in the "bad row" format (JSON).
output.bad.throttledBackoffPolicy.minBackoff (since 3.4.1)	Same as output.good.throttledBackoffPolicy.minBackoff for failed events in the "bad row" format (JSON).
output.bad.throttledBackoffPolicy.maxBackoff (since 3.4.1)	Same as output.good.throttledBackoffPolicy.maxBackoff for failed events in the "bad row" format (JSON).
output.bad.recordLimit	Same as output.good.recordLimit for failed events in the "bad row" format (JSON).
output.bad.customEndpoint	Same as output.good.customEndpoint for failed events in the "bad row" format (JSON).
output.pii.streamName	Optional. E.g. pii. Should be used in conjunction with the PII pseudonymization enrichment. When configured, enables an extra output stream for writing a pii_transformation event.
output.pii.region	Same as output.good.region for pii events.
output.pii.partitionKey	Same as output.good.partitionKey for pii events.
output.pii.backoffPolicy.minBackoff	Same as output.good.backoffPolicy.minBackoff for pii events.
output.pii.backoffPolicy.maxBackoff	Same as output.good.backoffPolicy.maxBackoff for pii events.
output.pii.backoffPolicy.maxRetries	Same as output.good.backoffPolicy.maxRetries for pii events.
output.pii.throttledBackoffPolicy.minBackoff (since 3.4.1)	Same as output.good.throttledBackoffPolicy.minBackoff for pii events.
output.pii.throttledBackoffPolicy.maxBackoff (since 3.4.1)	Same as output.good.throttledBackoffPolicy.maxBackoff for pii events.
output.pii.recordLimit	Same as output.good.recordLimit for pii events.
output.pii.customEndpoint	Same as output.good.customEndpoint for pii events.

enrich-kafka

A minimal configuration file can be found on the Github repo, as well as a comprehensive one.

parameter	description
input.topicName	Required. Name of the Kafka topic to read collector payloads from.
input.bootstrapServers	Required. A list of host:port pairs to use for establishing the initial connection to the Kafka cluster
input.consumerConf	Optional. Kafka consumer configuration. See the docs for all properties.
output.good.topicName	Required. Name of the Kafka topic to write to
output.good.bootstrapServers	Required. A list of host:port pairs to use for establishing the initial connection to the Kafka cluster
output.good.producerConf	Optional. Kafka producer configuration. See the docs for all properties
output.good.partitionKey	Optional. Enriched event field to use as Kafka partition key
output.good.headers	Optional. Enriched event fields to add as Kafka record headers
output.incomplete.topicName	Optional. Name of the Kafka topic that will receive the failed events (same format as the enriched events)
output.incomplete.bootstrapServers	Optional. A list of host:port pairs to use for establishing the initial connection to the Kafka cluster
output.incomplete.producerConf	Optional. Kafka producer configuration. See the docs for all properties
output.bad.topicName	Optional. Name of the Kafka topic that will receive the failed events in the “bad row” format (JSON
output.bad.bootstrapServers	Optional. A list of host:port pairs to use for establishing the initial connection to the Kafka cluster
output.bad.producerConf	Optional. Kafka producer configuration. See the docs for all properties
output.pii.topicName	Optional. Name of the Kafka topic to write to
output.pii.bootstrapServers	Optional. A list of host:port pairs to use for establishing the initial connection to the Kafka cluster
output.pii.producerConf	Optional. Kafka producer configuration. See the docs for all properties
output.pii.partitionKey	Optional. Enriched event field to use as Kafka partition key
output.pii.headers	Optional. Enriched event fields to add as Kafka record headers
blobStorage.s3 (since 4.0.0)	Optional. Set to true if S3 client should be initialized to download enrichments assets.
blobStorage.gcs (since 4.0.0)	Optional. Set to true if GCS client should be initialized to download enrichments assets.
blobStorage.azureStorage (since 4.0.0)	Optional. Azure Blob Storage client configuration. ABS client won't be enabled if it isn't given.
blobStorage.azureStorage.accounts (since 4.0.0)	Array of accounts to download from Azure Blob Storage.

Example values for the Azure storage accounts :

• { "name": "storageAccount1"}: public account with no auth
• { "name": "storageAccount2", "auth": { "type": "default"} }: private account using default auth chain
• { "name": "storageAccount3", "auth": { "type": "sas", "value": "tokenValue"}}: private account using SAS token auth

enrich-nsq

A minimal configuration file can be found on the Github repo, as well as a comprehensive one.

parameter	description
input.topic	Required. Name of the NSQ topic with the collector payloads.
input.lookupHost	Required. The host name of NSQ lookup application.
input.lookupPort	Required. The port number of NSQ lookup application.
input.channel	Optional. Default: collector-payloads-channel. Name of the NSQ channel used to retrieve collector payloads.
output.good.topic	Required. Name of the NSQ topic that will receive the enriched events.
output.good.nsqdHost	Required. The host name of nsqd application.
output.good.nsqdPort	Required. The port number of nsqd application.
output.incomplete.topic	Required. Name of the NSQ topic that will receive the failed events (same format as the enriched events).
output.incomplete.nsqdHost	Required. The host name of nsqd application.
output.incomplete.nsqdPort	Required. The port number of nsqd application.
output.bad.topic	Required. Name of the NSQ topic that will receive the failed events in the "bad row" format (JSON).
output.bad.nsqdHost	Required. The host name of nsqd application.
output.bad.nsqdPort	Required. The port number of nsqd application.
output.pii.topic	Optional. Name of the NSQ topic that will receive the pii events.
output.pii.nsqdHost	Optional. The host name of nsqd application.
output.pii.nsqdPort	Optional. The port number of nsqd application.

Enriched events validation against atomic schema

Enriched events are expected to match atomic schema. However, until 3.0.0, it was never checked that the enriched events emitted by enrich were valid. If an event is not valid against atomic schema, a failed event should be emitted instead of the enriched event. However, this is a breaking change, and we want to give some time to users to adapt, in case today they are working downstream with enriched events that are not valid against atomic. For this reason, this new validation was added as a feature that can be deactivated like that:

"featureFlags": {
  "acceptInvalid": true
}

In this case, enriched events that are not valid against atomic schema will still be emitted as before, so that Enrich 3.0.0 can be fully backward compatible. It will be possible to know if the new validation would have had an impact by 2 ways:

1. A new metric invalid_enriched has been introduced. It reports the number of enriched events that were not valid against atomic schema. As the other metrics, it can be seen on stdout and/or StatsD.
2. Each time there is an enriched event invalid against atomic schema, a line will be logged with the failed event (add -Dorg.slf4j.simpleLogger.log.InvalidEnriched=debug to the JAVA_OPTS to see it).

If acceptInvalid is set to false, a failed event will be emitted instead of the enriched event in case it's not valid against atomic schema.

When we'll know that all our customers don't have any invalid enriched events any more, we'll remove the feature flags and it will be impossible to emit invalid enriched events.

Since 4.0.0, it is possible to configure the lengths of the atomic fields, below is an example:

{
  ...
  # Optional. Configuration section for various validation-oriented settings.
  "validation": {
    # Optional. Configuration for custom maximum atomic fields (strings) length.
    # Map-like structure with keys being field names and values being their max allowed length
    "atomicFieldsLimits": {
        "app_id": 5
        "mkt_clickid": 100000
        # ...and any other 'atomic' field with custom limit
    }
  }
}

Enrichments

The list of the enrichments that can be configured can be found on this page.