Other Processes

Archival

Archiver

Any number of archiver processes can run simultaneously. All listen directly to the RabbitMQ dashboard.archivers worker queue and process traps in parallel, writing each to an elastic search index.

The index_prefix is diversified by appending the date string to derive the actual index name used.

Execution

An archiver process is started by executing the archiver shell script:

Usage: archiver [OPTIONS]

Options:
  --rmq_hostname TEXT     RabbitMQ hostname ["['test-dashboard-
                          storage01.geant.org', 'test-dashboard-
                          storage02.geant.org', 'test-dashboard-
                          storage03.geant.org']"]

  --es_hostname TEXT      Elasticsearch hostname [test-db-elk.geant.org]
  --es_index_prefix TEXT  Elasticsearch index name prefix [traps-]
  --es_doctype TEXT       Elasticsearch doc type [trap]
  --collection TEXT       collection pub/sub exchange name
                          ['dashboard.collection']

  --archivers TEXT        archiver worker ueue name
                          ['dashboard.archivers']
  --monitoring TEXT       monitoring requests exchange name
                          ['mon.requests']
  --username TEXT         RabbitMQ username ['dashboard']
  --password TEXT         RabbitMQ user password ['password']
  --vhost TEXT            RabbitMQ vhost ['/dashboard']
  --username TEXT         RabbitMQ username ['dashboard']
  --password TEXT         RabbitMQ user password ['password']
  --vhost TEXT            RabbitMQ vhost ['/dashboard']
  --watchdog INTEGER      watchdog check frequency in seconds
                          (no watchdog if unset)

  --timeout INTEGER       number of seconds without traps to indicate error
  --help                  Show this message and exit.

Notification Archiver

Any number of notification-archiver processes can run simultaneously. All listen directly to the RabbitMQ dashboard.notifiers.es_archiver worker queue and process alarm notifications sent to the dashboard.external.notifications exchange in parallel.

Each alarm notification is indexed to the Elasticsearch/Opensearch alarm-notifications index.

The notification-archiver process is configured via a JSON file formatted as follows:

_ARCHIVER_CONFIG_SCHEMA

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "definitions": {
    "hostnames": {
      "type": "array",
      "items": {
        "type": "string"
      },
      "minItems": 1
    },
    "rabbitmq": {
      "type": "object",
      "properties": {
        "hostnames": {
          "$ref": "#/definitions/hostnames"
        },
        "vhost": {
          "type": "string"
        },
        "username": {
          "type": "string"
        },
        "password": {
          "type": "string"
        },
        "archive": {
          "type": "object"
        }
      },
      "required": [
        "hostnames",
        "vhost",
        "username",
        "password"
      ],
      "additionalProperties": false
    },
    "elasticsearch": {
      "type": "object",
      "properties": {
        "hostnames": {
          "$ref": "#/definitions/hostnames"
        },
        "port": {
          "type": "integer"
        },
        "index-prefix": {
          "type": "string"
        },
        "index": {
          "type": "string"
        },
        "ssl": {
          "type": "boolean"
        },
        "username": {
          "type": "string"
        },
        "password": {
          "type": "string"
        }
      },
      "required": [
        "hostnames"
      ],
      "additionalProperties": false
    }
  },
  "type": "object",
  "properties": {
    "rmq": {
      "$ref": "#/definitions/rabbitmq"
    },
    "es": {
      "$ref": "#/definitions/elasticsearch"
    }
  },
  "required": [
    "rmq",
    "es"
  ],
  "additionalProperties": false
}

The archive element of the above configuration file must be formatted as follows:

ALARM_NOTIFICATIONS_RMQ_CONFIG

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "type": "object",
  "properties": {
    "notifications": {
      "type": "string"
    },
    "queue": {
      "type": "string"
    }
  },
  "required": [
    "notifications",
    "queue"
  ],
  "additionalProperties": false
}

Execution

A notification-archiver process is started by executing the notification-archiver shell script:

Usage: notification-archiver [OPTIONS]

Options:
  --config FILENAME  configuration filename  [required]
  --help             Show this message and exit.

State Archiver

Any number of state-archiver processes can run simultaneously. All consume messages (in round-robin fashion) sent by the correlator to the dashboard.alarms.broadcast exchange and by the isolation checker to the dashboard.isolation.broadcast exchange.

Each notification is indexed to the Elasticsearch/Opensearch dashboard-state index.

The state-archiver process is configured via a JSON file formatted as follows:

_ARCHIVER_CONFIG_SCHEMA

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "definitions": {
    "hostnames": {
      "type": "array",
      "items": {
        "type": "string"
      },
      "minItems": 1
    },
    "rabbitmq": {
      "type": "object",
      "properties": {
        "hostnames": {
          "$ref": "#/definitions/hostnames"
        },
        "vhost": {
          "type": "string"
        },
        "username": {
          "type": "string"
        },
        "password": {
          "type": "string"
        },
        "archive": {
          "type": "object"
        }
      },
      "required": [
        "hostnames",
        "vhost",
        "username",
        "password"
      ],
      "additionalProperties": false
    },
    "elasticsearch": {
      "type": "object",
      "properties": {
        "hostnames": {
          "$ref": "#/definitions/hostnames"
        },
        "port": {
          "type": "integer"
        },
        "index-prefix": {
          "type": "string"
        },
        "index": {
          "type": "string"
        },
        "ssl": {
          "type": "boolean"
        },
        "username": {
          "type": "string"
        },
        "password": {
          "type": "string"
        }
      },
      "required": [
        "hostnames"
      ],
      "additionalProperties": false
    }
  },
  "type": "object",
  "properties": {
    "rmq": {
      "$ref": "#/definitions/rabbitmq"
    },
    "es": {
      "$ref": "#/definitions/elasticsearch"
    }
  },
  "required": [
    "rmq",
    "es"
  ],
  "additionalProperties": false
}

The archive element of the above configuration file must be formatted as follows:

STATE_RMQ_CONFIG

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "definitions": {
    "exchange-queue": {
      "type": "object",
      "properties": {
        "exchange": {
          "type": "string"
        },
        "queue": {
          "type": "string"
        }
      },
      "required": [
        "exchange",
        "queue"
      ],
      "additionalProperties": false
    }
  },
  "type": "object",
  "properties": {
    "monitoring": {
      "type": "string"
    },
    "state": {
      "$ref": "#/definitions/exchange-queue"
    },
    "isolation": {
      "$ref": "#/definitions/exchange-queue"
    }
  },
  "required": [
    "monitoring",
    "state",
    "isolation"
  ],
  "additionalProperties": false
}

Execution

A state-archiver process is started by executing the state-archiver shell script:

Usage: state-archiver [OPTIONS]

Options:
  --config FILENAME  configuration filename  [required]
  --help             Show this message and exit.

Health Service

Dashboard Health Check Web Service

This is a Flask Web Service that returns the health of the critical Dashboard microservices.

This data is used in the status panel in the GUI.

Dashboard Health Check API

API Endpoints

/version

dashboard.health.routes.api.version()

Returns a json object with information about the module version.

The response will be formatted according to the following schema:

VERSION_SCHEMA

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "type": "object",
  "properties": {
    "api": {
      "type": "string",
      "pattern": "\\d+\\.\\d+"
    },
    "module": {
      "type": "string",
      "pattern": "\\d+\\.\\d+"
    }
  },
  "required": [
    "api",
    "module"
  ],
  "additionalProperties": false
}

Returns:: version json structure

/health

dashboard.health.routes.api.health()

Returns a json object with information about the module version.

The response will be formatted according to the following schema:

HEALTH_CHECK_RESPONSE_SCHEMA

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "definitions": {
    "process-status": {
      "type": "object",
      "properties": {
        "status": {
          "type": "string",
          "enum": [
            "healthy",
            "warning",
            "error"
          ]
        },
        "message": {
          "type": "string"
        },
        "timestamp": {
          "type": "integer"
        }
      },
      "required": [
        "status",
        "message",
        "timestamp"
      ],
      "additionalProperties": false
    }
  },
  "type": "object",
  "properties": {
    "correlator": {
      "$ref": "#/definitions/process-status"
    },
    "classifier": {
      "$ref": "#/definitions/process-status"
    },
    "collector": {
      "$ref": "#/definitions/process-status"
    },
    "inventory": {
      "$ref": "#/definitions/process-status"
    }
  },
  "required": [
    "correlator",
    "classifier",
    "collector",
    "inventory"
  ],
  "additionalProperties": false
}

Returns:: health json structure

Low-Level Check Endpoints

These endpoints are only useful for debugging and development. They’re low-level access to calling the health check functions directly.

/inventory-provider

dashboard.health.routes.check.inventory_provider()

Low-level endpoint for directly querying and returning inventory provider health.

Only intended for use during debugging/development.

/correlator

dashboard.health.routes.check.correlator()

Low-level endpoint for directly querying and returning correlator health.

Only intended for use during debugging/development.

/classifier

dashboard.health.routes.check.classifier()

Low-level endpoint for directly querying and returning classifier health.

Only intended for use during debugging/development.

/collector

dashboard.health.routes.check.collector()

Low-level endpoint for directly querying and returning collector health.

Only intended for use during debugging/development.

Common Endpoint Support Utitily Functions

dashboard.health.routes.common.after_request(rsp): generic function to do additional logging of requests & responses :param response: :return:

dashboard.health.routes.common.require_accepts_json(f)

used as a route handler decorator to return an error unless the request allows responses with type “application/json”

Parameters:: f – the function to be decorated
Returns:: the decorated function

RabbitMQ ping/pong utiltities

dashboard.health.rmq.get_channel(rmq_params, exchange_name, exchange_type='direct')

Create a channel to the RabbitMQ server using the configured connection parameters.

Parameters:

rmq_params – RabbitMQ connection parameters
exchange_name – Name of the exchange to use
exchange_type – Type of the exchange (default is ‘direct’)

Returns:

A context manager that yields the channel

dashboard.health.rmq.get(channel, queue, schema, timeout=2, stop_event=None)

Consume messages from a RabbitMQ queue and yields any that can be decoded and match the provide json schema.

Parameters:

channel – RabbitMQ channel to consume from
queue – Name of the queue to consume from
schema – JSON schema to validate messages against
timeout – Timeout for consuming messages (default is 2 seconds)
stop_event – Optional threading event to signal stopping the consumer (used only for testing)

Returns:

Yields messages that match the schema

dashboard.health.rmq._ping_pongs(channel, exchange, proc_type, schema, stop_event=None)

Send a PING message to the specified exchange and yield all PONG responses that match the provided JSON schema.

Parameters:

channel – RabbitMQ channel to use for publishing and consuming
exchange – Name of the exchange to broadcast to
proc_type – Type of process to ping (correlator, classifier, collector)
schema – JSON schema to validate PONG responses against
stop_event – Optional threading event to signal stopping the consumer (used only for testing)

Returns:

Yields all PONG responses that match the schema

dashboard.health.rmq.ping_proc_type(rmq_params, proc_type, schema, stop_event=None)

Creates a RabbitMQ channel and uses it to call _ping_pongs and return all matching PONG responses.

Parameters:

rmq_params – RabbitMQ connection parameters
proc_type – Type of process to ping (correlator, classifier, collector)
schema – JSON schema to validate PONG responses against
stop_event – Optional threading event to signal stopping the consumer (used only for testing)

Returns:

Yields all PONG responses that match the schema

Service Status Query utilities

dashboard.health.status.load_inventory_health(app_config)

picks a random endpoint from the configured inventory-version-uris list and queries the version endpoint

the health will indicate error if the version response reports an error condition or if the latest update has been pending for longer than the configured inventory_pending_error_threshold_s

the health status is set to warning if an update is in progress

Parameters:: app_config – the application config
Returns:: the health status of the inventory provider service

dashboard.health.status.load_correlator_health(app_config, stop_event=None)

pings all correlators and finds the leader. An error is returned if there’s no leader node found. _init_timestamp_health is then used to set the health of the service.

Parameters:

app_config – the application config
stop_event – the event to stop the consume loop (only used in tests)

Returns:

the health status of the correlator service

dashboard.health.status.load_classifier_health(app_config, stop_event=None)

pings all classifiers and uses _init_timestamp_health to set the health of the service. the status can also be set to WARNING if the number of classifiers is less than the configured expected_num_classifier_nodes

Parameters:

app_config – the application config
stop_event – the event to stop the consume loop (only used in tests)

Returns:

the health status of the overall classifier service

dashboard.health.status.load_collector_health(app_config, stop_event=None)

pings all collectors and uses _init_timestamp_health to set the health of the service

Parameters:

app_config – the application config
stop_event – the event to stop the consume loop (only used in tests)

Returns:

the health status of the overall collector service

dashboard.health.status._init_timestamp_health(thresholds, last_trap_ts)

compares last_trap_ts with the configured threshholds of trap_health_error_threshold_s and trap_health_warning_threshold_s. Returns a ProcessStatus object with the appropriate status and message.

Parameters:

thresholds – the configured health check thresholds
last_trap_ts – the last trap timestamp

Returns:

the health status of the service

App Environment Setup

dashboard.health.environment.setup_logging()

set up logging using the configured filename

if LOGGING_CONFIG is defined in the environment, use the contents: as the logging configuration, otherwise use _LOGGING_DEFAULT_CONFIG

dashboard.health.environment.setup_sentry(sentry_config)

set up sentry instrumentation

Parameters:: sentry_config – the app config ‘sentry’ element