things-search.conf

ditto {
  service-name = "things-search"
  mapping-strategy.implementation = "org.eclipse.ditto.thingsearch.api.ThingSearchMappingStrategies"

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

  mongodb {

    database = "searchDB"
    database = ${?MONGO_DB_DATABASE}

    pool {
      minSize = 0
      minSize = ${?MONGO_DB_CONNECTION_MIN_POOL_SIZE}
      maxSize = 1000
      maxSize = ${?MONGO_DB_CONNECTION_POOL_SIZE}
      maxWaitTime = 30s
      maxWaitTime = ${?MONGO_DB_CONNECTION_POOL_WAIT_TIME}
      jmxListenerEnabled = false
      jmxListenerEnabled = ${?MONGO_DB_CONNECTION_POOL_JMX_LISTENER_ENABLED}
    }

    breaker {
      # defines ater how many failures the circuit breaker should open
      maxFailures = 5
      maxFailures = ${?BREAKER_MAXFAILURES}
      timeout {
        # MongoDB Timeouts causing the circuitBreaker to open - "0s" disables timeouts opening the breaker
        call = 5s
        call = ${?BREAKER_TIMEOUT}
        # after this time in "Open" state, the cicuitBreaker is "Half-opened" again
        reset = 10s
        reset = ${?BREAKER_RESET}
      }
    }

    monitoring {
      commands = true
      commands = ${?MONGODB_MONITORING_COMMANDS_ENABLED}
      connection-pool = true
      commands = ${?MONGODB_MONITORING_CONNECTION_POOL_ENABLED}
    }

  }


  signal-enrichment {
    caching-signal-enrichment-facade.provider =  ${?THINGS_SEARCH_CACHING_SIGNAL_ENRICHMENT_PROVIDER}
  }

  things-search {
    query-criteria-validator.implementation = ${?QUERY_CRITERIA_VALIDATOR_IMPLEMENTATION}
    search-update-mapper.implementation = ${?SEARCH_UPDATE_MAPPER_IMPLEMENTATION}
    search-update-observer.implementation = ${?SEARCH_UPDATE_OBSERVER_IMPLEMENTATION}
    mongo-hints-by-namespace = ${?MONGO_HINTS_BY_NAMESPACE}

    index-initialization {
      # indices should be created within this application
      enabled = true
      enabled = ${?INDEX_INITIALIZATION_ENABLED}
    }

    updater {
      max-idle-time = 25h
      max-idle-time = ${?ACTIVITY_CHECK_INTERVAL}

      event-processing-active = true
      event-processing-active = ${?EVENT_PROCESSING_ACTIVE}

      # how often to poll shard region for state updates
      sharding-state-poll-interval = 15s
      sharding-state-poll-interval = ${?SHARDING_STATE_POLL_INTERVAL}

      force-update-probability = 0.00
      force-update-probability = ${?FORCE_UPDATE_PROBABILITY}

      force-update-after-start-timeout = 5m
      force-update-after-start-timeout = ${?FORCE_UPDATE_AFTER_START_TIMEOUT}

      background-sync {
        enabled = true
        enabled = ${?BACKGROUND_SYNC_ENABLED}

        quiet-period = 8m
        quiet-period = ${?BACKGROUND_SYNC_QUIET_PERIOD}

        idle-timeout = 5m
        idle-timeout = ${?BACKGROUND_SYNC_IDLE_TIMEOUT}

        tolerance-window = 20m
        tolerance-window = ${?BACKGROUND_SYNC_TOLERANCE_WINDOW}

        policy-ask-timeout = 10s
        policy-ask-timeout = ${?BACKGROUND_SYNC_POLICY_ASK_TIMEOUT}

        keep {
          events = 50
          events = ${?BACKGROUND_SYNC_KEEP_EVENTS}
        }

        throttle {
          throughput = 100
          throughput = ${?BACKGROUND_SYNC_THROTTLE_THROUGHPUT}

          period = 10s
          period = ${?BACKGROUND_SYCN_THROTTLE_PERIOD}
        }

        # handle failures/stalling/expired cursors
        min-backoff = 1s
        min-backoff = ${?BACKGROUND_SYNC_MIN_BACKOFF}

        max-backoff = 2m
        max-backoff = ${?BACKGROUND_SYNC_MAX_BACKOFF}

        # give up stream resumption after about 6 hours = 180 * 120s
        max-restarts = 180
        max-restarts = ${?BACKGROUND_SYNC_MAX_RESTARTS}

        # assume upstream healthy if no error happened for this long
        recovery = 5m
        recovery = ${?BACKGROUND_SYNC_RECOCVERY}
      }

      stream {
        # arrays bigger than this are not indexed
        max-array-size = 0
        max-array-size = ${?THINGS_SEARCH_UPDATER_STREAM_MAX_ARRAY_SIZE}

        # minimum delay between event dumps must be at least 1s
        write-interval = 1s
        write-interval = ${?THINGS_SEARCH_UPDATER_STREAM_WRITE_INTERVAL}

        # configuration for retrieval of policies/things via sharding
        ask-with-retry {
          ask-timeout = 5s
          ask-timeout = ${?THINGS_SEARCH_UPDATER_STREAM_ASK_TIMEOUT}

          # one of: OFF, NO_DELAY, FIXED_DELAY, BACKOFF_DELAY
          retry-strategy = BACKOFF_DELAY
          retry-strategy = ${?THINGS_SEARCH_UPDATER_STREAM_RETRY_STRATEGY}

          retry-attempts = 3
          retry-attempts = ${?THINGS_SEARCH_UPDATER_STREAM_RETRY_ATTEMPTS}

          fixed-delay = 5s
          fixed-delay = ${?THINGS_SEARCH_UPDATER_STREAM_RETRY_FIXED_DELAY}

          backoff-delay {
            min = 100ms
            max = 10s
            # must be between 0.0 and 1.0:
            random-factor = 0.5
          }
        }

        # retrieval of things and policy-enforcers
        retrieval {
          # upper bound of parallel SudoRetrieveThing commands (by extension, parallel loads of policy enforcer cache)
          parallelism = 16
          parallelism = ${?THINGS_SEARCH_UPDATER_STREAM_RETRIEVAL_PARALLELISM}

          # back-offs in case of failure
          exponential-backoff {
            min = 1s
            max = 2m
            random-factor = 2.0
          }
        }

        # writing into the persistence
        persistence {
          # how many bulk writes to request in parallel; must be a power of 2
          parallelism = 2
          parallelism = ${?THINGS_SEARCH_UPDATER_STREAM_PERSISTENCE_PARALLELISM}

          # how many write operations to perform in one bulk
          max-bulk-size = 250
          max-bulk-size = ${?THINGS_SEARCH_UPDATER_STREAM_PERSISTENCE_MAX_BULK_SIZE}

          # how long to wait after DB acknowledgement before sending "search-persisted" acknowledgement
          ack-delay = 0s
          ack-delay = ${?THINGS_SEARCH_UPDATER_STREAM_PERSISTENCE_ACK_DELAY}

          # mongoDB write concern to use for search updates requiring acknowledgement
          with-acks-writeConcern = journaled
          with-acks-writeConcern = ${?THINGS_SEARCH_UPDATER_STREAM_PERSISTENCE_WITH_ACKS_WRITE_CONCERN}

          # backoffs in case of failure
          exponential-backoff {
            min = 1s
            max = 2m
            random-factor = 2.0
          }
        }

        cache {
          # name of the dispatcher to run async cache loaders, which do not block threads
          dispatcher = "policy-enforcer-cache-dispatcher"

          # delay before retrying a cache query if the cached value is out of date
          retry-delay = 1s
          retry-delay = ${?THINGS_SEARCH_UPDATER_STREAM_CACHE_RETRY_DELAY}

          # how many enforcers to cache
          maximum-size = 20000
          maximum-size = ${?THINGS_SEARCH_UPDATER_STREAM_CACHE_SIZE}

          # lifetime of a cached enforcer. set very high because entries are invalidated lazily
          expire-after-write = 2h
          expire-after-write = ${?THINGS_SEARCH_UPDATER_STREAM_CACHE_EXPIRY}

          expire-after-access = 30m
          expire-after-access = ${?THINGS_SEARCH_UPDATER_STREAM_CACHE_EXPIRY_AFTER_ACCESS}
        }

      }

      persistence {
        # read preference is one of: primary, primaryPreferred, secondary, secondaryPreferred, nearest
        readPreference = ${ditto.mongodb.options.readPreference}
        readPreference = ${?UPDATER_PERSISTENCE_MONGO_DB_READ_PREFERENCE}

        # read concern is one of: default, local, majority, linearizable, snapshot, available
        readConcern = ${ditto.mongodb.options.readConcern}
        readConcern = ${?UPDATER_PERSISTENCE_MONGO_DB_READ_CONCERN}
      }
    }
  }
}

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

    roles = [
      "things-search",
      "blocked-namespaces-aware",
      "thing-event-aware"
    ]
  }
}

search-dispatcher {
  # one thread per query and a dedicated thread for the search actor
  type = PinnedDispatcher
  executor = "thread-pool-executor"
}

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
}

policy-enforcer-cache-dispatcher {
  type = "Dispatcher"
  executor = "thread-pool-executor"
  thread-pool-executor {
    keep-alive-time = 60s
    fixed-pool-size = off
    max-pool-size-max = 256
    max-pool-size-max = ${?CACHE_DISPATCHER_POOL_SIZE_MAX}
    max-pool-size-max = ${?POLICY_ENFORCER_CACHE_DISPATCHER_POOL_SIZE_MAX}
  }
}

search-updater-dispatcher {
  type = "Dispatcher"
  executor = "thread-pool-executor"
  thread-pool-executor {
    keep-alive-time = 60s
    fixed-pool-size = off
    max-pool-size-max = 256
    max-pool-size-max = ${?SEARCH_UPDATER_DISPATCHER_POOL_SIZE_MAX}
  }
}

include "things-search-extension.conf"