manager.asyncapi.yaml

# vim: set et ts=2 sw=2:

id: urn:rpc:metricq:manager
asyncapi: '2.0.0'

tags:
  - name: "management"
    description: >-
      RPCs processed by management clients
  - name: "direct"
    description: >-
      RPCs sent to a specific client directly
  - name: "broadcast"
    description: >-
      RPCs sent to all clients on the network at once
  - name: "source"
    description: >-
      RPCs used by Source instances
  - name: "sink"
    description: >-
      RPCs used by Sink instances
  - name: "transformer"
    description: >-
      RPCs used by Transformer instances
  - name: "database"
    description: >-
      RPCs used by Database instances
  - name: "history"
    description: >-
      RPCs used by history clients

info:
  title: MetricQ Management Protocol
  version: '0.1.0'
  description: |
    Clients on the MetricQ network are managed via an RPC protocol, where
    messages are encoded in a JSON format.
    A request consists of a JSON object with a property `"function"`, identifying
    the action to perform.
    Arguments are passed by specifying properties on the request object.
    Replies are arbitrary JSON objects.
    If a request could not be processed successfully, the reply has a single
    property `"error"` containing an error message as a string.

    RPCs are grouped into three categories and are tagged as such:

    * **`management`**:
      Management RPCs are sent to an exchange called `metricq.management`, management
      clients subscribe to messages on this queue.
      Replies from management clients arrive on a dedicated *management queue*.
    * **`broadcast`**:
      Broadcast RPCs are sent to an exchange called `metric.broadcast`.
      All clients will receive such an request simultaneously and are expected to reply.
    * **`direct`**:
      Some RPCs are sent directly into a client's RPC queue, bypassing any exchange.

servers:
  production:
    url: "metricq.zih.tu-dresden.de"
    protocol: ampq

defaultContentType: application/json

channels:
  get_metrics:
    description: >-
      Query the network for available metrics
    publish:
      tags:
        - name: "management"
      message:
        description: "Request information about available metrics"
        bindings:
          amqp:
            replyTo:
              type: string
        correlationId:
          location: $message.header#/correlation_id
        payload:
          $ref: "#/components/schemas/rpcFunction"
          properties:
            function:
              const: "get_metrics"
              $ref: "#/components/schemas/rpcFunctionName"
            format:
              description: "Whether to return an array of metric names or an object mapping metric names to their metadata"
              type: string
              enum: ["array", "object"]
              default: "array"
            historic:
              description: >-
                If given, limit returned metrics to either historic (`true`) or non-historic (`false`) metrics.
                If omitted, no filter is applied.
              type: boolean
            limit:
              description: >-
                Number of matched metrics to include in response.
                If omitted, all matching metrics are returned.
              type: integer
              minimum: 0
            hidden:
              description: >-
                If given, limit returned metrics to either hidden (`true`) or not hidden (`false`) metrics.
                If omitted, this filter is not applied.
              type: boolean
          oneOf:
            - properties:
                selector:
                  $ref: "#/components/schemas/metricSelector"
            - properties:
                prefix:
                  type: string
                  description: "A prefix matching the start of a metric name"
          examples:
            - function: "get_metrics"
              selector: "taurus\\.taurusi\\d{4}\\.cpu\\d\\+.power"
              limit: 20
            - function: "get_metrics"
              prefix: "taurus"
              historic: true
    subscribe:
      tags:
        - name: "management"
      message:
        name: "get_metrics"
        description: "A list of matching metrics, with optional metadata"
        payload:
          type: object
          properties:
            metrics:
                $ref: "#/components/schemas/metricCollection"

  get_configs:
    description: >-
      Retrieve the configuration object for a set of clients
    publish:
      tags:
        - name: "management"
      message:
        name: "get_configs"
        description: "Retrieve configuration objects for matching clients"
        payload:
          $ref: "#/components/schemas/rpcFunction"
          properties:
            function:
              const: "get_configs"
              $ref: "#/components/schemas/rpcFunctionName"
            selector:
              description: "A selector matching clients by their token"
              oneOf:
                - type: array
                  description: "A list of client tokens"
                  items:
                    type: string
                - type: string
                  format: regex
                  description: "A regular expression matching a set of client tokens"
    subscribe:
      tags:
        - name: "management"
      message:
        name: "get_configs"
        description: "A mapping of client names to their configuration objects"
        payload:
          type: object
          additionalProperties:
            $ref: "#/components/schemas/configObject"
          examples:
            - "source-py-dummy":
                rate: 0.5

  sink.register:
    description: >-
      Register a new sink on the network
    publish:
      tags:
        - name: "sink"
        - name: "management"
      message:
        name: "sink.register"
        payload:
          $ref: "#/components/schemas/rpcFunction"
          properties:
            function:
              const: "sink.register"
              $ref: "#/components/schemas/rpcFunctionName"
          examples:
            - function: "sink.register"
    subscribe:
      tags:
        - name: "sink"
        - name: "management"
      message:
        name: "sink.register"
        description: >-
          On success, this contains the information necessary to start consuming metrics from the network.
          Optionally, a configuration object for this sink is returned.
        payload:
          type: object
          properties:
            dataServerAddress:
              $ref: "#/components/schemas/dataServerAddress"
            config:
              $ref: "#/components/schemas/configObject"


  sink.subscribe:
    description: >-
      Request to receive data points for a set of metrics.
    publish:
      tags:
        - name: "sink"
        - name: "management"
      message:
        name: "sink.subscribe"
        payload:
          $ref: "#/components/schemas/rpcFunction"
          properties:
            function:
              const: "sink.subscribe"
              $ref: "#/components/schemas/rpcFunctionName"
            metrics:
              $ref: "#/components/schemas/metricList"
              description: "A list of metrics to subscribe to."
            dataQueue:
              $ref: "#/components/schemas/dataQueue"
              description: "The AMQP queue on which data points should arrive."
            expires:
              $ref: "#/components/schemas/seconds"
              description: >-
                The number of seconds after which the data queue should expire if not connected to a sink.
                If omitted, a default value is set.
              exclusiveMinimum: 0
            metadata:
              $ref: "#/components/schemas/metadataArgument"
          required:
            - "function"
            - "metrics"
          examples:
            - function: "sink.subscribe"
              metrics: ["building-F.room-A14.temperature.celcius"]
              metadata: true
    subscribe:
      tags:
        - name: "sink"
        - name: "management"
      message:
        description: >-
          After successful subscription, this contains the broker address and
          queue on which data points for the requested metrics arrive.
          The metrics in question are returned, optionally with their metadata attached.
        payload:
          $ref: "#/components/schemas/subscribeResponse"

  sink.unsubscribe:
    publish:
      tags:
        - name: "sink"
        - name: "management"
      message:
        name: "sink.unsubscribe"
        description: "Subscribe to receive data points for a set of metrics"
        payload:
          title: "sink.unsubscribe"
          description: "Stop receiving data points for a set of metrics"
          properties:
            dataQueue:
              $ref: "#/components/schemas/dataQueue"
              description: >-
                The AMQP queue at which data points should no longer be sent.
            metrics:
              $ref: "#/components/schemas/metricList"
            end:
              type: boolean
              description: >-
                If `true`, the manager sends a `type="end"` message after unbinding the routing keys to indicate that no more message will arrive at this queue.
              default: true
          required:
            - "dataQueue"
            - "metrics"
    subscribe:
      tags:
        - name: "sink"
        - name: "management"
      message:
        description: >-
          An acknowledgement of unsubscription
        payload:
          type: object
          properties:
            dataServerAddress:
                $ref: "#/components/schemas/dataServerAddress"
            metrics:
                description: "The collection of metrics this client is no longer subscribed to"
                $ref: "#/components/schemas/metricCollection"

  sink.release:
    publish:
      tags:
        - name: "sink"
        - name: "management"
      message:
        name: "sink.release"
        title: "sink.release"
        description: "Release a data queue"
        payload:
          $ref: "#/components/schemas/rpcFunction"
          properties:
            function:
              const: "sink.release"
              $ref: "#/components/schemas/rpcFunctionName"
            dataQueue:
              $ref: "#/components/schemas/dataQueue"
              description: "The data queue to be released"
    subscribe:
      tags:
        - name: "sink"
        - name: "management"
      message:
        payload:
          type: object
          description: "Acknowledgement of successful release"

  source.register:
    publish:
      tags:
        - name: "source"
        - name: "management"
      message:
        name: "source.register"
        summary: "Register a source on the network"
        payload:
          $ref: "#/components/schemas/rpcFunction"
          properties:
            function:
              const: "source.register"
              $ref: "#/components/schemas/rpcFunctionName"
    subscribe:
      tags:
        - name: "source"
        - name: "management"
      message:
        payload:
          type: object
          properties:
            dataServerAddress:
              $ref: "#/components/schemas/dataServerAddress"
            dataExchange:
              $ref: "#/components/schemas/dataExchange"
            config:
              $ref: "#/components/schemas/configObject"

  source.declare_metrics:
    publish:
      tags:
        - name: "source"
        - name: "management"
      message:
        name: "source.declare_metrics"
        summary: "Declare a list of metrics which this source provides"
        payload:
          type: object
          $ref: "#/components/schemas/rpcFunction"
          properties:
            function:
              const: "source.declare_metrics"
              $ref: "#/components/schemas/rpcFunctionName"
            metrics:
              $ref: "#/components/schemas/metricCollection"
          required:
            - "metrics"
    subscribe:
      tags:
        - name: "source"
        - name: "management"
      message:
        payload:
          $ref: "#/components/schemas/declareMetricsResponse"

  transformer.register:
    publish:
      tags:
        - name: "transformer"
        - name: "management"
      message:
        name: "transformer.register"
        summary: "Register a transformer on the network"
        payload:
          $ref: "#/components/schemas/rpcFunction"
          properties:
            function:
              const: "transformer.register"
              $ref: "#/components/schemas/rpcFunctionName"
    subscribe:
      tags:
        - name: "transformer"
        - name: "management"
      message:
        payload:
          type: object
          properties:
            dataExchange:
              $ref: "#/components/schemas/dataExchange"
            config:
              $ref: "#/components/schemas/configObject"

  transformer.declare_metrics:
    publish:
      tags:
        - name: "transformer"
        - name: "management"
      message:
        name: "transformer.declare_metrics"
        summary: "Declare a list of metrics which this transformer provides"
        payload:
          type: object
          $ref: "#/components/schemas/rpcFunction"
          properties:
            function:
              const: "transformer.declare_metrics"
              $ref: "#/components/schemas/rpcFunctionName"
            metrics:
              $ref: "#/components/schemas/metricCollection"
          required:
            - "metrics"
    subscribe:
      tags:
        - name: "transformer"
        - name: "management"
      message:
        payload:
          $ref: "#/components/schemas/declareMetricsResponse"

  transformer.subscribe:
    publish:
      tags:
        - name: "transformer"
        - name: "management"
      message:
        name: "transformer.subscribe"
        title: "transformer.subscribe"
        description: "Subscribe to receive data points for a set of metrics"
        payload:
          $ref: "#/components/schemas/rpcFunction"
          properties:
            function:
              const: "transformer.subscribe"
              $ref: "#/components/schemas/rpcFunctionName"
            metrics:
              $ref: "#/components/schemas/metricList"
            metadata:
              $ref: "#/components/schemas/metadataArgument"
          required:
            - "function"
            - "metrics"
    subscribe:
      tags:
        - name: "transformer"
        - name: "management"
      message:
        title: "transformer.subscribe"
        description: >-
          `transformer.subscribe` response
        payload:
          $ref: "#/components/schemas/subscribeResponse"

  db.register:
    description: >-
      Register a database on the network.
    publish:
      tags:
        - name: "database"
        - name: "management"
      message:
        name: "db.register"
        payload:
          $ref: "#/components/schemas/rpcFunction"
          properties:
            function:
              const: "db.register"
              $ref: "#/components/schemas/rpcFunctionName"
    subscribe:
      tags:
        - name: "database"
        - name: "management"
      message:
        payload:
          type: object
          properties:
            dataServerAddress:
              $ref: "#/components/schemas/dataServerAddress"
            dataExchange:
              $ref: "#/components/schemas/dataExchange"
            dataQueue:
              $ref: "#/components/schemas/dataQueue"
              description: "Metrics to be saved in the DB arrive on this queue"
            historyQueue:
              $ref: "#/components/schemas/historyQueue"
            config:
              $ref: "#/components/schemas/configObject"

  db.subscribe:
    description: >-
      Request to receive data points for metrics to be saved in the DB.
    publish:
      tags:
        - name: "database"
        - name: "management"
      message:
        name: "db.subscribe"
        payload:
          $ref: "#/components/schemas/rpcFunction"
          properties:
            function:
              const: "db.subscribe"
              $ref: "#/components/schemas/rpcFunctionName"
            metrics:
              description: >-
                A list of metrics to save, possibly with custom names for history requests.
              oneOf:
                - type: array
                  description: "A list of metrics to save in the DB"
                  items:
                    $ref: "#/components/schemas/metricName"
                - type: array
                  description: >-
                    A list of metrics to save, with custom names for history requests.

                    Queue bindings for the `dataQueue` will use `input`,
                    while queue bindings for the `historyQueue` will use `name`.
                  items:
                    type: object
                    properties:
                      name:
                        $ref: "#/components/schemas/metricName"
                      input:
                        $ref: "#/components/schemas/metricName"
                    required:
                      - name
                      - input
            metadata:
              $ref: "#/components/schemas/metadataArgument"
          required:
            - "function"
            - "metrics"
          examples:
            - function: "db.subscribe"
              metrics: ["building-F.room-A14.temperature.celcius"]
              metadata: true
            - function: "db.subscribe"
              metrics:
                - name: "history-name.foo-metric"
                  input: "test.foo-metric"
    subscribe:
      tags:
        - name: "database"
        - name: "management"
      message:
        description: >-
          After successful subscription, this contains the broker address and
          queue on which data points for the requested metrics arrive.
          The metrics in question are returned, optionally with their metadata attached.
        payload:
          $ref: "#/components/schemas/subscribeResponse"

  history.register:
    description: >-
      Register a history client on the network.
    publish:
      tags:
        - name: "history"
        - name: "management"
      message:
        name: "history.register"
        payload:
          $ref: "#/components/schemas/rpcFunction"
          properties:
            function:
              const: "history.register"
              $ref: "#/components/schemas/rpcFunctionName"
    subscribe:
      tags:
        - name: "history"
        - name: "management"
      message:
        payload:
          type: object
          properties:
            dataServerAddress:
              $ref: "#/components/schemas/dataServerAddress"
            historyExchange:
              type: string
              examples:
                - "metricq.history"
            historyQueue:
              $ref: "#/components/schemas/historyQueue"
            config:
              $ref: "#/components/schemas/configObject"

  discover:
    publish:
      tags:
        - name: "broadcast"
      message:
        name: "discover"
        summary: "A discovery request sent to all clients online"
        payload:
          $ref: "#/components/schemas/rpcFunction"
          properties:
            function:
              const: "discover"
              $ref: "#/components/schemas/rpcFunctionName"
    subscribe:
      tags:
        - name: "broadcast"
      message:
        name: "discover"
        summary: "A discovery response including operational information of a client"
        payload:
          properties:
            alive:
              type: boolean
              default: true
            currentTime:
              type: string
              format: date-time
              description: "Current time on this host"
            startingTime:
              type: string
              format: date-time
              description: "When this client was started and first connected to the network"
            uptime:
              type: integer
              format: nanoseconds
              description: >-
                Number of nanoseconds since startup of this client.

                If missing, assume this is the difference between `currentTime` and `startingTime`.
            metricqVersion:
              type: string
              description: >-
                The version of the MetricQ client library that this client uses to communicate with the network.

                This value is intended for human consumption and its format should not be relied opon.
                Nonetheless, a format of _<library-name>/<version>_ is encouraged.
              examples:
                - "metricq-python/1.1.4"
                - "metricq-cpp/r501.ge944379"
            pythonVersion:
              type: string
              description: >-
                The version of the running Python interpreter, if the client uses the MetricQ Python bindings.
              examples:
                - "3.9.2"
            hostname:
              type: string
              description: >-
                The hostname identifying the machine that this client runs on.
              examples:
                - "bahka"
                - "192.0.2.42"

  config:
    publish:
      tags:
        - name: "direct"
      message:
        name: "config"
        summary: >-
          Provide an updated configuration object to a client
        payload:
          $ref: "#/components/schemas/rpcFunction"
          properties:
            function:
              const: "config"
              $ref: "#/components/schemas/rpcFunctionName"
          additionalProperties: true
    subscribe:
      tags:
        - name: "direct"
      message:
        name: "config"
        summary: >-
          Configuration acknowledgement
        payload:
          type: object
          description: "Acknowledgement that configuration has been received successfully"



components:
  schemas:
    rpcFunction:
      type: object
      properties:
        function:
          $ref: "#/components/schemas/rpcFunctionName"
      required:
        - "function"

    rpcFunctionName:
      type: string
      title: "RPC function name"
      description: "A string identifying the requested RPC function"
      pattern: "([a-z_]+\\.)+[a-z_]+"

    metricName:
      type: string
      pattern: "([a-zA-Z][a-zA-Z0-9_]+\\.)+[a-zA-Z][a-zA-Z0-9_]+"
      examples:
        - "elab.bahka.power"
        - "elab.ariel.package0.dram.current"
        - "taurus.taurusi4101.cpu0.power"

    metricList:
      type: array
      description: "A list of metric names"
      items:
        $ref: "#/components/schemas/metricName"
      uniqueItems: true

    metricsWithMetadata:
      type: object
      description: "A mapping of metric names to metadata objects"
      additionalProperties:
        $ref: "#/components/schemas/metadata"
      examples:
        - elab.bahka.power:
            rate: 20
            scope: "last"
            unit: "W"

    metricCollection:
      description: "A collection of metric names, with optional metadata attached"
      oneOf:
        - $ref: "#/components/schemas/metricList"
        - $ref: "#/components/schemas/metricsWithMetadata"

    metricSelector:
      description: "A selector matching a set of metrics, either by regex or explicit listing"
      oneOf:
        - type: string
          format: regex
          description: "A regular expression matching a set of metric names"
          examples:
            - "taurus\\.taurusi\\d{4}\\.cpu\\d\\+.power"
        - $ref: "#/components/schemas/metricList"

    dataQueue:
      type: string
      format: "uri"
      examples:
        - "amqp://user:pass@localhost/vhost"
        - "amqps://metricq.example.com:50505/"

    historyQueue:
      type: string
      format: "uri"
      description: "Requests for historic data arrive on this queue"
      examples:
        - "amqp://user:pass@localhost/history"
        - "amqps://metricq.example.com:50505/"

    seconds:
      type: number
      format: seconds

    metadata:
      description: "A metadata object"
      type: object
      properties:
        unit:
          type: string
          title: "Unit symbol"
          description: >-
            The unit symbol according to SI, such as `"V"` for voltage or `"s"` for time.


            If necessary, the SI prefix is included, e.g. `"kV"`, `"ms"`, or `"kiB/s"`.
            _Dimensionless_ quantities such as counts use either an empty string (`""`) or the appropriate symbol (e.g. "rad").
          examples:
            - "V"
            - "kV"
            - "s"
            - "kiB/s"
            - "rad"
            - ""
        rate:
          type: integer
          description: >-
            The rate in `Sa/s` (samples per second) or `Hz` at which new values of this metric are typically produced.
            In practice, a source does not necessarily need to produce values exactly at this rate, but the value given should be good estimate.
        scope:
          enum: ["last", "next", "point"]
          description: >-
            The temporal validity of a single measured value.


            A scope of `"last"` indicates that a measured value was valid _up to now_, `"next"` that it is valid _from now on_.
            A scope of `"point"` means that any measurement is only valid at exactly the moment it was taken.
        quantity:
          type: string
          description: >-
            The name of the measured quantity.


            Ideally, this is the same as the last segment of the metric name.
          examples:
            - "power"
            - "temperature"
        description:
          type: string
          description: >-
            A human-readable description of the measurement quantity
      examples:
        - rate: 0.1

    configObject:
      description: "A client configuration object"
      type: object
      examples:
        - reportNeighboursDogActivity: true
          metricSourceChannels:
            "kittens.front-flap.opened.count": "automation.doors.front.catflap.triggered#cumulative"
            "kittens.water-bowl.1.volume": "automation.volume-sensors[5]"
        - clientSpecificOption: 42
          monitorKittenHealth: true


    dataServerAddress:
      type: string
      format: url
      description: "URL pointing to the AMQP broker where data queues are handled"
      examples:
        - "amqp://user:pass@localhost/data-vhost"
        - "amqps://metricq.example.com:50505/"

    dataExchange:
      type: string
      description: "Name of the exchange to which data points should be published"
      examples:
        - "metricq.data"

    declareMetricsResponse:
      type: object
      description: "Acknowledgement that metrics have been declared successfully"

    metadataArgument:
      type: boolean
      default: false
      description: >-
        Whether to include metadata for each metric in the reponse.


        If set to `false`, the result is an array of metric names,
        if set to `true`, an object mapping metric names to metadata objects is returned.

    subscribeResponse:
      type: object
      properties:
        dataServerAddress:
            $ref: "#/components/schemas/dataServerAddress"
        dataQueue:
            $ref: "#/components/schemas/dataQueue"
            description: >-
              The AMQP queue at which data points for the subscribed metrics will arrive.
        metrics:
            $ref: "#/components/schemas/metricCollection"