policies.conf

ditto {
  service-name = "policies"
  mapping-strategy.implementation = "org.eclipse.ditto.policies.api.PoliciesMappingStrategies"
  extensions {
    pre-enforcer-provider {
      extension-class = org.eclipse.ditto.policies.enforcement.pre.PreEnforcerProvider
      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.policies.service.enforcement.pre.ModifyToCreatePolicyTransformer" // always keep this as first transformer in order to guarantee that all following transformers know that the command is creating a policy instead of modifying it
    ]
    snapshot-adapter = {
      extension-class = "org.eclipse.ditto.policies.service.persistence.serializer.PolicyMongoSnapshotAdapter"
    }
  }

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

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

  policies {

    policy {
      # the configuration to which duration the "expiry" of a Policy Subject should be rounded up.
      # For example:
      # * configured to "1s": a received "expiry" is rounded up to the next full second
      # * configured to "30s": a received "expiry" is rounded up to the next half minute
      # * configured to "1h": a received "expiry" is rounded up to the next full hour
      # * configured to "12h": a received "expiry" is rounded up to the next half day
      # * configured to "1d": a received "expiry" is rounded up to the next full day
      # * configured to "15d": a received "expiry" is rounded up to the next half month
      subject-expiry-granularity = 1h
      subject-expiry-granularity = ${?POLICY_SUBJECT_EXPIRY_GRANULARITY}

      # To which duration the notify-before duration of each subject-expiry is rounded up.
      # Minimum value: 1ms
      subject-deletion-announcement-granularity = 1m
      subject-deletion-announcement-granularity = ${?POLICY_SUBJECT_DELETION_ANNOUNCEMENT_GRANULARITY}

      subject-id-resolver = "org.eclipse.ditto.policies.service.persistence.actors.resolvers.DefaultSubjectIdFromActionResolver"
      subject-id-resolver = ${?POLICY_SUBJECT_ID_RESOLVER}

      activity-check {
        # the interval of how long to keep an "inactive" Policy in memory:
        inactive-interval = 2h
        inactive-interval = ${?POLICY_ACTIVITY_CHECK_INTERVAL}

        # the interval of how long to keep a deleted Policy in memory
        deleted-interval = 5m
        deleted-interval = ${?POLICY_ACTIVITY_CHECK_DELETED_INTERVAL}
      }

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

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

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

      announcement {
        grace-period = 4h
        grace-period = ${?POLICY_ANNOUNCEMENT_GRACE_PERIOD}

        max-timeout = 60s
        max-timeout = ${?POLICY_ANNOUNCEMENT_MAX_TIMEOUT}

        enable-announcements-when-deleted = true
        enable-announcements-when-deleted = ${?POLICY_ENABLE_ANNOUNCEMENTS_WHEN_DELETED}

        exponential-backoff {
          # minimum backoff for announcement redelivery
          min = 1s
          min = ${?POLICY_ANNOUNCEMENT_BACKOFF_MIN}
          # maximum backoff for announcement
          max = 1h
          max = ${?POLICY_ANNOUNCEMENT_BACKOFF_MAX}
          # random factor for exponential back-off
          # next-backoff = previous-backoff * (1.5 +  random-factor * rand(0, 0.5))
          random-factor = 1.0
          random-factor = ${?POLICY_ANNOUNCEMENT_BACKOFF_RANDOM_FACTOR}
        }
      }

      cleanup {
        enabled = true
        enabled = ${?CLEANUP_ENABLED}

        quiet-period = 5m
        quiet-period = ${?CLEANUP_QUIET_PERIOD}

        interval = 10s
        interval = ${?CLEANUP_INTERVAL}

        timer-threshold = 150ms
        timer-threshold = ${?CLEANUP_TIMER_THRESHOLD}

        credits-per-batch = 3
        credits-per-batch = ${?CLEANUP_CREDITS_PER_BATCH}

        reads-per-query = 100
        reads-per-query = ${?CLEANUP_READS_PER_QUERY}

        writes-per-credit = 100
        writes-per-credit = ${?CLEANUP_WRITES_PER_CREDIT}

        delete-final-deleted-snapshot = false
        delete-final-deleted-snapshot = ${?CLEANUP_DELETE_FINAL_DELETED_SNAPSHOT}
      }
    }

    persistence-ping {
      # journal tag to query to find our which PolicyPersistenceActors to ping (always keep alive)
      journal-tag = "always-alive"
      journal-tag = ${?POLICIES_PERSISTENCE_PING_JOURNAL_TAG}
      # initial delay for pinging the PolicyPersistenceActors after the PersistencePingActor has been started.
      initial-delay = 0s
      initial-delay = ${?POLICIES_PERSISTENCE_PING_INITIAL_DELAY}
      # interval for trying to ping all "always-alive" PolicyPersistenceActors
      interval = 10m
      interval = ${?POLICIES_PERSISTENCE_PING_INTERVAL}
      # how many events to read in one query
      read-journal-batch-size = 500
      read-journal-batch-size = ${?POLICIES_PERSISTENCE_PING_READ_JOURNAL_BATCH_SIZE}

      # one of:
      #  TAGS  Elements will be ordered by their tags.
      #  ID    Elements will be ordered by their document ID.
      streaming-order = ID
      streaming-order = ${?POLICIES_PERSISTENCE_PING_STREAMING_ORDER}

      # used to throttle pinging of PolicyPersistenceActors, so that not all PolicyPersistenceActors are recovered at the same time
      rate {
        frequency = 1s
        frequency = ${?POLICIES_PERSISTENCE_PING_RATE_FREQUENCY}
        entities = 1
        entities = ${?POLICIES_PERSISTENCE_PING_RATE_ENTITIES}
      }
    }
  }
}

akka {
  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 = [
      "policies",
      "policy-announcement-aware",
      "ddata-aware",
      "blocked-namespaces-aware"
    ]
  }

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

akka-contrib-mongodb-persistence-policies-journal {
  class = "akka.contrib.persistence.mongodb.MongoJournal"
  plugin-dispatcher = "policy-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 = "policies_journal"
    journal-index = "policies_journal_index"

    realtime-collection = "policies_realtime"
    metadata-collection = "policies_metadata"
  }

  event-adapters {
    mongodbobject = "org.eclipse.ditto.policies.service.persistence.serializer.DefaultPolicyMongoEventAdapter"
  }

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

akka-contrib-mongodb-persistence-policies-snapshots {
  class = "akka.contrib.persistence.mongodb.MongoSnapshots"
  plugin-dispatcher = "policy-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 = "policies_snaps"
    snaps-index = "policies_snaps_index"
  }
}

policy-journal-persistence-dispatcher {
  type = Dispatcher
  # which mailbox to use
  mailbox-type = "org.eclipse.ditto.policies.service.persistence.actors.PolicyPersistenceActorMailbox"
  mailbox-capacity = 100
  executor = "fork-join-executor"
  fork-join-executor {
    parallelism-min = 4
    parallelism-factor = 3.0
    parallelism-max = 32
    parallelism-max = ${?DEFAULT_DISPATCHER_PARALLELISM_MAX}
  }
  throughput = 5
}

policy-snaps-persistence-dispatcher {
  type = Dispatcher
  # which mailbox to use
  mailbox-type = "org.eclipse.ditto.policies.service.persistence.actors.PolicyPersistenceActorMailbox"
  mailbox-capacity = 100
  executor = "fork-join-executor"
  fork-join-executor {
    parallelism-min = 4
    parallelism-factor = 3.0
    parallelism-max = 32
    parallelism-max = ${?DEFAULT_DISPATCHER_PARALLELISM_MAX}
  }
  throughput = 5
}

blocked-namespaces-dispatcher {
  type = Dispatcher
  executor = "fork-join-executor"
  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
}

akka.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_JOURNAL_MAX_SIZE}
    }
  }
}

include "policies-extension.conf"