things.conf

ditto {
  service-name = "things"
  mapping-strategy.implementation = "org.eclipse.ditto.things.api.ThingsMappingStrategies"

  extensions {
    pre-enforcer-provider.extension-config.pre-enforcers = [
      "org.eclipse.ditto.policies.enforcement.pre.BlockedNamespacePreEnforcer",
      "org.eclipse.ditto.policies.enforcement.pre.CommandWithOptionalEntityPreEnforcer",
      "org.eclipse.ditto.policies.enforcement.pre.CreationRestrictionPreEnforcer"
    ]
    signal-transformers-provider.extension-config.signal-transformers = [
      "org.eclipse.ditto.things.service.enforcement.pre.ModifyToCreateThingTransformer", // always keep this as first transformer in order to guarantee that all following transformers know that the command is creating a thing instead of modifying it
      "org.eclipse.ditto.things.service.signaltransformation.placeholdersubstitution.ThingsPlaceholderSubstitution"
    ]
    snapshot-adapter = {
      extension-class = "org.eclipse.ditto.things.service.persistence.serializer.ThingMongoSnapshotAdapter"
      extension-config {
        thing-snapshot-taken-event-publishing-enabled = true
        thing-snapshot-taken-event-publishing-enabled = ${?THING_SNAPSHOT_TAKEN_EVENT_PUBLISHING_ENABLED}
      }
    }
  }

  mongodb {
    database = "things"
    database = ${?MONGO_DB_DATABASE}

    read-journal {
      # additional index which speeds up background (e.g. cleanup) aggregation queries
      # seems to be required for MongoDB version >=6, otherwise a lot of read Disk IOPS are needed
      should-create-additional-snapshot-aggregation-index-pid-id = false
      should-create-additional-snapshot-aggregation-index-pid-id = ${?MONGODB_READ_JOURNAL_SHOULD_CREATE_ADDITIONAL_SNAPSHOT_AGGREGATION_INDEX_PID_ID}

      # additional index which speeds up background (e.g. cleanup) aggregation queries
      # seems to be required for MongoDB version >=6, otherwise a lot of read Disk IOPS are needed
      should-create-additional-snapshot-aggregation-index-pid-sn = false
      should-create-additional-snapshot-aggregation-index-pid-sn = ${?MONGODB_READ_JOURNAL_SHOULD_CREATE_ADDITIONAL_SNAPSHOT_AGGREGATION_INDEX_PID_SN}

      # additional index which speeds up background (e.g. cleanup) aggregation queries
      # seems to be required for MongoDB version >=6, otherwise a lot of read Disk IOPS are needed
      should-create-additional-snapshot-aggregation-index-pid-sn-id = false
      should-create-additional-snapshot-aggregation-index-pid-sn-id = ${?MONGODB_READ_JOURNAL_SHOULD_CREATE_ADDITIONAL_SNAPSHOT_AGGREGATION_INDEX_PID_SN_ID}

      hint-name-filterPidsThatDoesntContainTagInNewestEntry = null
      hint-name-filterPidsThatDoesntContainTagInNewestEntry = ${?MONGODB_READ_JOURNAL_HINT_NAME_FILTER_PIDS_THAT_DOESNT_CONTAIN_TAG_IN_NEWEST_ENTRY}

      hint-name-listLatestJournalEntries = null
      hint-name-listLatestJournalEntries = ${?MONGODB_READ_JOURNAL_HINT_NAME_LIST_LATEST_JOURNAL_ENTRIES}

      hint-name-listNewestActiveSnapshotsByBatchPidId = null
      hint-name-listNewestActiveSnapshotsByBatchPidId = ${?MONGODB_READ_JOURNAL_HINT_NAME_LIST_NEWEST_ACTIVE_SNAPSHOT_BY_BATCH_PID_ID}

      hint-name-listNewestActiveSnapshotsByBatchPid = null
      hint-name-listNewestActiveSnapshotsByBatchPid = ${?MONGODB_READ_JOURNAL_HINT_NAME_LIST_NEWEST_ACTIVE_SNAPSHOT_BY_BATCH_PID}

      hint-name-listNewestActiveSnapshotsByBatchId = null
      hint-name-listNewestActiveSnapshotsByBatchId = ${?MONGODB_READ_JOURNAL_HINT_NAME_LIST_NEWEST_ACTIVE_SNAPSHOT_BY_BATCH_ID}
    }
  }

  persistence.operations.delay-after-persistence-actor-shutdown = 5s
  persistence.operations.delay-after-persistence-actor-shutdown = ${?DELAY_AFTER_PERSISTENCE_ACTOR_SHUTDOWN}

  things {

    # Logs for all incoming messages minimal information to enable message tracing throughout the system
    log-incoming-messages = true
    log-incoming-messages = ${?LOG_INCOMING_MESSAGES}

    thing {

      shutdown-timeout = 3s
      shutdown-timeout = ${?THING_SHUTDOWN_TIMEOUT}

      activity-check {
        # the interval of how long to keep an "inactive" Thing in memory:
        inactive-interval = 2h
        inactive-interval = ${?THING_ACTIVITY_CHECK_INTERVAL} # may be overridden with this environment variable

        # the interval of how long to keep an "inactive" deleted Thing in memory:
        deleted-interval = 5m
        deleted-interval = ${?THING_ACTIVITY_CHECK_DELETED_INTERVAL}
      }

      snapshot {
        # the interval when to do snapshot for a Thing which had changes to it
        interval = 15m
        interval = ${?THING_SNAPSHOT_INTERVAL} # may be overridden with this environment variable

        # the threshold after how many changes to a Thing to do a snapshot
        threshold = 500
        threshold = ${?THING_SNAPSHOT_THRESHOLD} # may be overridden with this environment variable
      }

      event {
        # define the DittoHeaders to persist when persisting events to the journal
        # those can e.g. be retrieved as additional "audit log" information when accessing a historical thing revision
        historical-headers-to-persist = [
          #"ditto-originator"  # who (user-subject/connection-pre-auth-subject) issued the event
          #"correlation-id"
          #"ditto-origin"      # which WS session or connection issued the event
          #"origin"            # the HTTP origin header
          #"user-agent"        # the HTTP user-agent header
        ]
        historical-headers-to-persist = ${?THING_EVENT_HISTORICAL_HEADERS_TO_PERSIST}
      }

      supervisor {
        exponential-backoff {
          min = 1s
          max = 120s
          random-factor = 1.0
          corrupted-receive-timeout = 600s
        }

        local-ask {
          timeout = 5s
          timeout = ${?THINGS_SUPERVISOR_LOCAL_ASK_TIMEOUT}
        }
      }

      cleanup {
        # enabled configures whether background cleanup is enabled or not
        # If enabled, stale "snapshot" and "journal" entries will be cleaned up from the MongoDB by a background process
        enabled = true
        enabled = ${?CLEANUP_ENABLED}

        # history-retention-duration configures the duration of how long to "keep" events and snapshots before being
        # allowed to remove them in scope of cleanup.
        # If this e.g. is set to 30d - then effectively an event history of 30 days would be available via the read
        # journal.
        history-retention-duration = 3d
        history-retention-duration = ${?CLEANUP_HISTORY_RETENTION_DURATION}

        # quiet-period defines how long to stay in a state where the background cleanup is not yet started
        # Applies after:
        # - starting the service
        # - each "completed" background cleanup run (all entities were cleaned up)
        quiet-period = 5m
        quiet-period = ${?CLEANUP_QUIET_PERIOD}

        # interval configures how often a "credit decision" is made.
        # The background cleanup works with a credit system and does only generate new "cleanup credits" if the MongoDB
        # currently has capacity to do cleanups.
        interval = 3s
        interval = ${?CLEANUP_INTERVAL}

        # timer-threshold configures the maximum database latency to give out credit for cleanup actions.
        # If write operations to the MongoDB within the last `interval` had a `max` value greater to the configured
        # threshold, no new cleanup credits will be issued for the next `interval`.
        # Which throttles cleanup when MongoDB is currently under heavy (write) load.
        timer-threshold = 150ms
        timer-threshold = ${?CLEANUP_TIMER_THRESHOLD}

        # credits-per-batch configures how many "cleanup credits" should be generated per `interval` as long as the
        # write operations to the MongoDB are less than the configured `timer-threshold`.
        # Limits the rate of cleanup actions to this many per credit decision interval.
        # One credit means that the "journal" and "snapshot" entries of one entity are cleaned up each `interval`.
        credits-per-batch = 3
        credits-per-batch = ${?CLEANUP_CREDITS_PER_BATCH}

        # reads-per-query configures the number of snapshots to scan per MongoDB query.
        # Configuring this to high values will reduce the need to query MongoDB too often - it should however be aligned
        # with the amount of "cleanup credits" issued per `interval` - in order to avoid long running queries.
        reads-per-query = 100
        reads-per-query = ${?CLEANUP_READS_PER_QUERY}

        # writes-per-credit configures the number of documents to delete for each credit.
        # If for example one entity would have 1000 journal entries to cleanup, a `writes-per-credit` of 100 would lead
        # to 10 delete operations performed against MongoDB.
        writes-per-credit = 100
        writes-per-credit = ${?CLEANUP_WRITES_PER_CREDIT}

        # delete-final-deleted-snapshot configures whether for a deleted entity, the final snapshot (containing the
        # "deleted" information) should be deleted or not.
        # If the final snapshot is not deleted, re-creating the entity will cause that the recreated entity starts with
        # a revision number 1 higher than the previously deleted entity. If the final snapshot is deleted as well,
        # recreation of an entity with the same ID will lead to revisionNumber=1 after its recreation.
        delete-final-deleted-snapshot = false
        delete-final-deleted-snapshot = ${?CLEANUP_DELETE_FINAL_DELETED_SNAPSHOT}
      }
    }

    wot {
      http {
        # proxy config for retrieving WoT (Web of Things) ThingModels via HTTP
        proxy {
          enabled = false
          enabled = ${?THINGS_WOT_THING_MODEL_HTTP_PROXY_ENABLED}

          hostname = ${?THINGS_WOT_THING_MODEL_HTTP_PROXY_HOST}
          port = ${?THINGS_WOT_THING_MODEL_HTTP_PROXY_PORT}
          username = ${?THINGS_WOT_THING_MODEL_HTTP_PROXY_USERNAME}
          password = ${?THINGS_WOT_THING_MODEL_HTTP_PROXY_PASSWORD}
        }
      }

      cache {
        # how many thing models to cache
        maximum-size = 1000
        maximum-size = ${?THINGS_WOT_THING_MODEL_CACHE_SIZE}

        # prolonged on each cache access by that duration
        expire-after-access = 1d
        expire-after-access = ${?THINGS_WOT_THING_MODEL_CACHE_EXPIRE_AFTER_ACCESS}
      }

      tm-based-creation {
        thing {
          skeleton-creation-enabled = true
          skeleton-creation-enabled = ${?THINGS_WOT_TM_BASED_CREATION_THING_SKELETON_CREATION_ENABLED}

          generate-defaults-for-optional-properties = false
          generate-defaults-for-optional-properties = ${?THINGS_WOT_TM_BASED_CREATION_THING_GENERATE_DEFAULTS_FOR_OPTIONAL_PROPERTIES}

          throw-exception-on-wot-errors = true
          throw-exception-on-wot-errors = ${?THINGS_WOT_TM_BASED_CREATION_THING_THROW_EXCEPTION_ON_WOT_ERRORS}
        }

        feature {
          skeleton-creation-enabled = true
          skeleton-creation-enabled = ${?THINGS_WOT_TM_BASED_CREATION_FEATURE_SKELETON_CREATION_ENABLED}

          generate-defaults-for-optional-properties = false
          generate-defaults-for-optional-properties = ${?THINGS_WOT_TM_BASED_CREATION_FEATURE_GENERATE_DEFAULTS_FOR_OPTIONAL_PROPERTIES}

          throw-exception-on-wot-errors = true
          throw-exception-on-wot-errors = ${?THINGS_WOT_TM_BASED_CREATION_FEATURE_THROW_EXCEPTION_ON_WOT_ERRORS}
        }
      }

      to-thing-description {
        base-prefix = "http://localhost:8080"
        base-prefix = ${?THINGS_WOT_TO_THING_DESCRIPTION_BASE_PREFIX}

        # OAuth 2.0 example using Google:
        # json-template {
        #   "securityDefinitions": {
        #     "oauth2_google_sc": {
        #       "scheme": "oauth2",
        #       "authorization": "https://accounts.google.com/o/oauth2/v2/auth",
        #       "token": "https://oauth2.googleapis.com/token",
        #       "scopes": [
        #         "openid"
        #       ],
        #       "flow": "code"
        #     }
        #   },
        #   "security": "oauth2_google_sc"
        #   "support": "https://www.eclipse.dev/ditto/"
        # }

        json-template {
          "securityDefinitions": {
            "basic_sc": {
              "scheme": "basic",
              "in": "header"
            }
          },
          "security": "basic_sc"
          "support": "https://www.eclipse.dev/ditto/"
        }

        placeholders {
          # add arbitrary placeholders to be resolved, e.g.:
          # FOO = "bar"
          # TM_OPTIONAL = [
          #  "/properties/status",
          #  "/actions/toggle"
          # ]
        }

        add-created = true
        add-created = ${?THINGS_WOT_TO_THING_DESCRIPTION_ADD_CREATED}

        add-modified = false
        add-modified = ${?THINGS_WOT_TO_THING_DESCRIPTION_ADD_MODIFIED}
      }
    }
  }
}


pekko {
  cluster {
    sharding {
      role = ${ditto.service-name}

      passivation {
        # don't passivate shards by default as Ditto AbstractPersistenceActor decides that on its own -
        # default is "default-idle-strategy" with "idle-entity.timeout = 120s":
        strategy = "off"
      }
    }

    roles = [
      "things",
      "thing-event-aware",
      "live-signal-aware",
      "acks-aware",
      "blocked-namespaces-aware"
    ]
  }

  persistence {
    journal.auto-start-journals = [
      "pekko-contrib-mongodb-persistence-things-journal"
    ]
    snapshot-store.auto-start-snapshot-stores = [
      "pekko-contrib-mongodb-persistence-things-snapshots"
    ]
  }

  management.health-checks.liveness-checks {
    subsystem-health = "org.eclipse.ditto.internal.utils.health.SubsystemHealthCheck"
  }

}

pekko-contrib-mongodb-persistence-things-journal {
  class = "pekko.contrib.persistence.mongodb.MongoJournal"
  plugin-dispatcher = "thing-journal-persistence-dispatcher"

  circuit-breaker {
    max-failures = 5 # if an exception during persisting an event/snapshot occurs this often -- a successful write resets the counter
    max-failures = ${?BREAKER_MAXTRIES}
    call-timeout = 5s # MongoDB Timeouts causing the circuitBreaker to open
    call-timeout = ${?BREAKER_TIMEOUT}
    reset-timeout = 3s # after this time in "Open" state, the cicuitBreaker is "Half-opened" again
    reset-timeout = ${?BREAKER_RESET}
  }

  overrides {
    journal-collection = "things_journal"
    journal-index = "things_journal_index"

    realtime-collection = "things_realtime"
    metadata-collection = "things_metadata"
    metadata-index = "things_metadata_index"

  }

  event-adapters {
    mongodbobject = "org.eclipse.ditto.things.service.persistence.serializer.ThingMongoEventAdapter"
  }

  event-adapter-bindings {
    "org.eclipse.ditto.things.model.signals.events.ThingEvent" = mongodbobject
    "org.bson.BsonValue" = mongodbobject
  }
}

pekko-contrib-mongodb-persistence-things-journal-read {
  class = "pekko.contrib.persistence.mongodb.MongoReadJournal"
  plugin-dispatcher = "thing-journal-persistence-dispatcher"

  overrides {
    journal-collection = "things_journal"
    journal-index = "things_journal_index"
    realtime-collection = "things_realtime"
    metadata-collection = "things_metadata"
    metadata-index = "things_metadata_index"
  }
}

pekko-contrib-mongodb-persistence-things-snapshots {
  class = "pekko.contrib.persistence.mongodb.MongoSnapshots"
  plugin-dispatcher = "thing-snaps-persistence-dispatcher"

  circuit-breaker {
    max-failures = 5 # if an exception during persisting an event/snapshot occurs this often -- a successful write resets the counter
    max-failures = ${?SNAPSHOT_BREAKER_MAXTRIES}
    call-timeout = 10s # MongoDB Timeouts causing the circuitBreaker to open
    call-timeout = ${?SNAPSHOT_BREAKER_TIMEOUT}
    reset-timeout = 6s # after this time in "Open" state, the cicuitBreaker is "Half-opened" again
    reset-timeout = ${?SNAPSHOT_BREAKER_RESET}
  }

  overrides {
    snaps-collection = "things_snaps"
    snaps-index = "things_snaps_index"
  }
}

thing-journal-persistence-dispatcher {
  type = Dispatcher
  # which mailbox to use
  mailbox-type = "org.eclipse.ditto.things.service.persistence.actors.ThingPersistenceActorMailbox"
  mailbox-capacity = 100
  mailbox-capacity = ${?THING_PERSISTENCE_ACTOR_MAILBOX_SIZE}
  executor = "org.eclipse.ditto.internal.utils.metrics.executor.InstrumentedForkJoinExecutorServiceConfigurator"
  fork-join-executor {
    parallelism-min = 4
    parallelism-factor = 3.0
    parallelism-max = 32
    parallelism-max = ${?DEFAULT_DISPATCHER_PARALLELISM_MAX}
  }
  throughput = 5
}

thing-snaps-persistence-dispatcher {
  type = Dispatcher
  # which mailbox to use
  mailbox-type = "org.eclipse.ditto.things.service.persistence.actors.ThingPersistenceActorMailbox"
  mailbox-capacity = 100
  mailbox-capacity = ${?THING_PERSISTENCE_ACTOR_MAILBOX_SIZE}
  executor = "org.eclipse.ditto.internal.utils.metrics.executor.InstrumentedForkJoinExecutorServiceConfigurator"
  fork-join-executor {
    parallelism-min = 4
    parallelism-factor = 3.0
    parallelism-max = 32
    parallelism-max = ${?DEFAULT_DISPATCHER_PARALLELISM_MAX}
  }
  throughput = 5
}

wot-dispatcher {
  type = Dispatcher
  executor = "org.eclipse.ditto.internal.utils.metrics.executor.InstrumentedThreadPoolExecutorServiceConfigurator"
}
wot-dispatcher-cache-loader {
  type = Dispatcher
  executor = "org.eclipse.ditto.internal.utils.metrics.executor.InstrumentedThreadPoolExecutorServiceConfigurator"
}

blocked-namespaces-dispatcher {
  type = Dispatcher
  executor = "org.eclipse.ditto.internal.utils.metrics.executor.InstrumentedForkJoinExecutorServiceConfigurator"
  fork-join-executor {
    # Min number of threads to cap factor-based parallelism number to
    parallelism-min = 4
    # Parallelism (threads) ... ceil(available processors * factor)
    parallelism-factor = 3.0
    # Max number of threads to cap factor-based parallelism number to
    parallelism-max = 32
    parallelism-max = ${?DEFAULT_DISPATCHER_PARALLELISM_MAX}
  }
  throughput = 5
}

pekko.contrib.persistence.mongodb.mongo {

  collection-cache {
    journal {
      class = "org.eclipse.ditto.internal.utils.persistence.mongo.DittoMongoCollectionCache"
      expire-after-write = 1h // check each collection every hour whether it was created correctly
      expire-after-write = ${?MONGO_COLLECTION_CACHE_JOURNAL_TTL}
      max-size = 1000 // namespace collections cached per node
      max-size = ${?MONGO_COLLECTION_CACHE_JOURNAL_MAX_SIZE}
    }

    snaps {
      class = "org.eclipse.ditto.internal.utils.persistence.mongo.DittoMongoCollectionCache"
      expire-after-write = 1h
      expire-after-write = ${?MONGO_COLLECTION_CACHE_SNAPS_TTL}
      max-size = 1000
      max-size = ${?MONGO_COLLECTION_CACHE_SNAPS_MAX_SIZE}
    }
  }
}

include "things-extension.conf"
include file("/opt/ditto/things-extension.conf")