Configuration

In order to configure the service, a set of environment variables are adopted for describing basic service needs, alongside a main configuration file.

Environment Variables

Variable	Description	Default
LOG_LEVEL	the maximum log level to emit. Accepted levels are trace|debug|info|warn|error	info
HTTP_PORT	the HTTP port on which kubernetes status routes and metrics are exposed	3000
CONFIGURATION_FOLDER	the filepath to the folder under which configuration file is located	<HOME>/.df/mongezium
OTEL_EXPORTER_OTLP_ENDPOINT	specify the OpenTelemetry OTLP endpoint where traces and metrics should be pushed. When not set, telemetry is not exported

info

Currently <HOME> value is set to /home/mongezium, which is based on how the service image is built.

Configuration File

The application needs a configuration file, named config.json, which respects the following JSON schema configuration.

Schema Viewer

Loading ....

Raw JSON Schema

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "MongeziumConfiguration",
  "type": "object",
  "properties": {
    "collections": {
      "description": "Describe which collection should be monitored for changes",
      "type": "array",
      "items": {
        "$ref": "#/definitions/CollectionConfiguration"
      }
    },
    "persistence": {
      "description": "Describe how to connect to the persistence layer, that is a MongoDB cluster",
      "allOf": [
        {
          "$ref": "#/definitions/PersistenceConfiguration"
        }
      ]
    },
    "stream": {
      "description": "Describe how to connect to the stream layer, that is a Kafka cluster",
      "allOf": [
        {
          "$ref": "#/definitions/StreamConfiguration"
        }
      ]
    }
  },
  "required": [
    "collections",
    "persistence",
    "stream"
  ],
  "definitions": {
    "CollectionConfiguration": {
      "description": "Contains the properties that describe which MongoDB collection should be monitored\nand how its changes should be processed.",
      "type": "object",
      "properties": {
        "topic": {
          "description": "Name of the Kafka topic onto which change events related to this collection will be produced",
          "type": "string"
        },
        "namespace": {
          "description": "MongoDB namespace representing a collection within a database. It is expected in the form `<database-name>.<collection-name>`.",
          "type": "string"
        },
        "snapshot": {
          "description": "Describe how the service should address the _snapshot_ operation, that is the whole collection content\nis loaded onto the Kafka topic before producing events related to changes occurring to the collection records.\n\n_Snapshot_ operations are usually launched the first time the service starts monitoring a collection,\nwhen the business requires, such as when a new clean state is needed onto the Kafka topic.\n\nBy default, _snapshot_ operation is carried only when no message has\never been written to the selected topic for this collection.",
          "default": "initial",
          "allOf": [
            {
              "$ref": "#/definitions/SnapshotMode"
            }
          ]
        },
        "tombstone": {
          "description": "Specifies whether for each DELETE event also a _Tombstone_ event is generated.\nA _Tombstone_ event is a Kafka message with a non-null key and a null payload. This\ntype of event enables deleting old messages not needed anymore when using the Kafka\n[log compaction](https://docs.confluent.io/kafka/design/log_compaction.html#compaction-enables-deletes) policy.",
          "type": "boolean",
          "default": false
        }
      },
      "required": [
        "topic",
        "namespace"
      ]
    },
    "SnapshotMode": {
      "description": "Describe when the snapshot operation should be carried out.",
      "oneOf": [
        {
          "description": "Execute snapshot operation when no message\nhas ever been produced to the topic or when no Change Stream resume token is found",
          "type": "string",
          "const": "when_needed"
        },
        {
          "description": "Execute snapshot operation only when no message\nhas ever been produced to the topic",
          "type": "string",
          "const": "initial"
        }
      ]
    },
    "PersistenceConfiguration": {
      "description": "Configuration of the persistence layer, that is MongoDB connection string.",
      "type": "object",
      "properties": {
        "url": {
          "description": "MongoDB connection string",
          "allOf": [
            {
              "$ref": "#/definitions/Secret"
            }
          ]
        },
        "appName": {
          "description": "Client application name, which is used to identify the application in MongoDB logs.",
          "type": [
            "string",
            "null"
          ]
        }
      },
      "required": [
        "url"
      ]
    },
    "Secret": {
      "examples": [
        "my-secret",
        {
          "type": "env",
          "key": "CUSTOM_ENV_VAR"
        },
        {
          "type": "env",
          "key": "CUSTOM_ENV_VAR",
          "encoding": "base64"
        },
        {
          "type": "file",
          "path": "/path/to/file"
        }
      ],
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "object",
          "properties": {
            "type": {
              "const": "env"
            },
            "key": {
              "type": "string"
            },
            "encoding": {
              "description": "Define which type of encoding the library supports when it needs to read the actual secret value.",
              "type": "string",
              "enum": [
                "base64"
              ]
            }
          },
          "required": [
            "type",
            "key"
          ]
        },
        {
          "type": "object",
          "properties": {
            "type": {
              "const": "file"
            },
            "key": {
              "type": "string"
            },
            "path": {
              "type": "string"
            },
            "encoding": {
              "description": "Define which type of encoding the library supports when it needs to read the actual secret value.",
              "type": "string",
              "enum": [
                "base64"
              ]
            }
          },
          "required": [
            "type",
            "path"
          ]
        }
      ]
    },
    "StreamConfiguration": {
      "description": "Configuration of the stream layer, that is Kafka clients.",
      "type": "object",
      "properties": {
        "consumer": {
          "description": "Configurations related to the Kafka Consumer employed to retrieve\nthe latest state produced by the service alongside previous change events.\n\n`bootstrap.servers` property MUST have the value of the corresponding property in the `producer` configuration.",
          "type": "object",
          "additional_properties": {
            "$ref": "#/definitions/Secret"
          },
          "required": [
            "bootstrap.servers"
          ]
        },
        "producer": {
          "description": "Configurations related to the Kafka Producer for sending to Kafka topics\nchange events occurring on the persistence layer.\n\n`bootstrap.servers` property MUST have the value of the corresponding property in the `consumer` configuration.",
          "type": "object",
          "additional_properties": {
            "description": "librdkafka producer config | https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md",
            "$ref": "#/definitions/Secret"
          },
          "required": [
            "bootstrap.servers"
          ]
        }
      },
      "required": [
        "consumer",
        "producer"
      ]
    }
  }
}

Example

{
  "$schema": "https://docs.mia-platform.eu/schemas/fast_data/mongezium.0.4.2.schema.json",
  "persistence": {
    "url": {
      "type": "file",
      "path": "/run/secrets/mongodb/url"
    }
  },
  "collections": [
    {
      "topic": "fd.topic-1",
      "snapshot": "when_needed",
      "namespace": "fd.collection-1"
    },
    {
      "topic": "fd.topic-2",
      "snapshot": "initial",
      "namespace": "fd.collection-2"
    },
    {
      "topic": "fd.topic-3",
      "snapshot": "when_needed",
      "namespace": "fd.collection-3"
    }
  ],
  "stream": {
    "consumer": {
      "bootstrap.servers": {
        "type": "file",
        "path": "/run/secrets/kafka/bootstrap.servers"
      },
      "group.id": "fd.mongezium"
    },
    "producer": {
      "bootstrap.servers": {
        "type": "file",
        "path": "/run/secrets/kafka/bootstrap.servers"
      },
      "message.timeout.ms": "5000",
      "compression.type": "snappy"
    }
  }
}

note

The raw JSON schema can also be found here.

In addition, Kafka configurations and MongoDB persistence properties support secret resolution.

Kubernetes

Resources

When the plugin is deployed on Kubernetes, it is advised to set its resources requests and limits. Here are provided which are the recommended ones, although they can be changed according to your needs:

Recommended

• requests:

  CPU: 250m 
  Memory: 50MB

• limits:

  CPU: 1000m 
  Memory: 150MB

Status Probes

The service exposes the liveness and readines status probes as HTTP endpoint, which helps Kubernetes when the service is successfully started and when it may need to be restarted.

The endpoints are:

• liveness probe: /-/healthz
• readiness probe: /-/ready