SevOne Data Publisher

1. Prepare a Kafka broker or a Pulsar broker to which SDP can publish data points. If you are new to Kafka or Pulsar, please refer to the following links for a quick setup.
   • Kafka: https://kafka.apache.org/quickstart
   • Pulsar: https://pulsar.apache.org/docs/2.10.x/getting-started-standalone/#start-pulsar-standalone
2. SSH into your SevOne NMS appliance or cluster as root and it is not necessary to use sudo with the commands. However, if you do not SSH in as root, you must precede your commands with sudo.

   $ ssh root@<NMS appliance>

3. Change directory to /config/sdp.

   $ cd /config/sdp

4. Copy example-config.yml file to config.yml.

   $ cp example-config.yml config.yml

   Important: For reference, the complete /config/sdp/example-config.yml file released with SevOne NMS can be found in section Appendix: Configuration File below.
5. Using the text editor of your choice, edit config.yml.

   Example

   $ vi config.yml

6. In the config.yml file,
   1. if you are using the Kafka broker,
      • search for bootstrap.servers. Populate the value with <your Kafka IP address>:<port number>.
   2. If you are using the Pulsar broker,
      • search for serviceUrl. Populate the value with pulsar://<your Pulsar IP address>:<port number>.
      • change output.default.type from kafka to pulsar.
7. Save config.yml file.
8. Start SDP.

   
   $ podman exec -it nms-nms-nms /bin/bash
   
   $ SevOne-act sdp enable

9. Wait for at most one-minute. You should see data points arriving to your Kafka / Pulsar broker.
10. Stop SDP.

    
    $ podman exec -it nms-nms-nms /bin/bash
    
    $ SevOne-act sdp disable

The following sections show how to configure various settings present in config.yml file.

SDP uses Kafka/Pulsar producers to publish data points to Kafka/Pulsar brokers. The producers are configurable as follows.

This feature enables you to republish historical data for a specified time period.

Script/usr/local/scripts/utilities/republish-historical-data.sh, is provided to start historical backfill. You must provide the start time and end time as parameters to the script in UNIX timestamp format. Once the historical data request is made, SevOne Data Publisher will collect all data points for each indicator present on the peer and start to republish them to historical brokers.

1. Execute the following command to start historical backfill.

   $ /usr/local/scripts/utilities/republish-historical-data.sh <start time> <end time>

   Example

   $ /usr/local/scripts/utilities/republish-historical-data.sh 1697026113.058791
   1697026113.058791
   =============================
   | Web Server Status |
   =============================
   | Http Enabled | True
   | Http Port | 8082
   | Https Enabled | False
   | Https Port | 8443
   ============================
   Http Server Enabled, the republish endpoint is http://localhost:8082/status/republish
   Send republishing request through http://localhost:8082/status/republish
   Started to republished data from 1697026113.058791 to 1697026113.058791

   Important: At a time only one republishing request can be executed. If multiple requests are made, you will get a message to wait for the previous request to complete.

   Example

   $ /usr/local/scripts/utilities/republish-historical-data.sh 1697026113.058791
   1697026113.058791
   =============================
   | Web Server Status |
   =============================
   | Http Enabled | True
   | Http Port | 8082
   | Https Enabled | False
   | Https Port | 8443
   ============================
   Http Server Enabled, the republish endpoint is http://localhost:8082/status/republish
   Send republishing request through http://localhost:8082/status/republish
   Can't Process: Republisher is already running, please wait for the previous request to
   complete

2. Historical backfill stops automatically after the request has completed. However, to forcefully stop the historical backfill, execute the command below.

   $ /usr/local/scripts/utilities/republish-historical-data.sh STOP

   Example

   $ /usr/local/scripts/utilities/republish-historical-data.sh STOP
   =============================
   | Web Server Status |
   =============================
   | Http Enabled | True
   | Http Port | 8082
   | Https Enabled | False
   | Https Port | 8443
   ============================
   Http Server Enabled, the republish endpoint is http://localhost:8082/status/republish
   Stop republishing through http://localhost:8082/status/republish/stop
   republishing has been stopped

SevOne Data Publisher supports the following GET endpoints.

Examples: API commands

Here are a few examples on how to obtain the output for the API commands listed in the table above.

Example# 1

$ curl -s http://10.128.11.132:8082/api/publishers | jq
{
    "default-hist-producer": {
        "dataPointsSentFailureCount": 14,
        "dataPointsSentSuccessCount": 7
    },
    "default-producer": {
        "dataPointsSentFailureCount": 1,
        "dataPointsSentSuccessCount": 5
    }
}

Example# 2

$ curl -s http://10.129.13.198:8082/api/filters | jq
{
    "excludeFilters": null,
    "includeFilters": [
        {
            "name": "Everything",
             "matchCount": 305
        }
    ]
}

Example# 3

$ curl -s http://10.129.13.198:8082/api/system/uptime | jq
"0h:16m:10s:175ms"

SevOne Data Publisher outputs the data in either avro or json formats.

Using a text editor of your choice, update the schema for the avro output in /config/sdp/schema.json file to include / exclude the following supported fields.

Example: /config/sdp/schema.json file

{
    "fields": [
        { "name": "deviceId", "type": "int"},
        { "name": "deviceName", "type": "string"},
        { "name": "deviceIp", "type": "string"},
        { "name": "peerId", "type": "int"},
        { "name": "objectId", "type": "int"},
        { "name": "objectName", "type": "string"},
        { "name": "objectDesc", "type": "string"},
        { "name": "pluginId", "type": "int"},
        { "name": "pluginName", "type": "string"},
        { "name": "indicatorId", "type": "int"},
        { "name": "indicatorName", "type": "string"},
        { "name": "format", "type": "int"},
        { "name": "value", "type": "string"},
        { "name": "time", "type": "double"},
        { "name": "clusterName", "type": "string"},
        { "name": "peerIp", "type": "string"},
        { "name": "objectType", "type": "string"},
        { "name": "units", "type": "string"}
    ]
    "name": "sevone_msg_schema",
    "type": "record"
}

From SevOne NMS, example-config.yml file is available from /config/sdp.

Example: /config/sdp/example-config.yml file

version: 1

log:
  # Accepted values: error, warn, info, debug
  level: info

cache:
  refreshPeriodSeconds: 1800

# Configure the SDP output format
sdp:
  outputFormat: avro
  clusterName: NMS
  includeDeviceOID: false
  schemaFile: /config/sdp/schema.json
  # number of workers for live data publishing. Defaults to 10 workers if unspecified
  workers: 10

# Filters for the SDP output
# Note: SDP only exports data from the local peer, so these filters only apply to local data.
include-filters:
  - name: default
    # Specify your filters as different elements in this array
    # by specifying an ID that you would like to be included.
    # A value of -1 is interpreted as any ID.

    # Each column in a given filter is combined with logical AND.
    # Multiple filters are combined with logical OR.
    rules:
      # Example: Include everything
      - { devGrpID: -1, devID: -1, objGrpID: -1, objID: -1, pluginID: -1 }
  - name: allowlist1
    rules:
      # Example: Include only devices 5 and 6
      - { devGrpID: -1, devID: 5, objGrpID: -1, objID: -1, pluginID: -1 }
      - { devGrpID: -1, devID: 6, objGrpID: -1, objID: -1, pluginID: -1 }

      # Example: Include device 5 only if it's in device group 2 and device 6
      # - { devGrpID: 2, devID: 5, objGrpID: -1, objID: -1, pluginID: -1 }
      # - { devGrpID: -1, devID: 6, objGrpID: -1, objID: -1, pluginID: -1 }

      # Example: Include only objects 2, 3, and 4 from device 5
      # - { devGrpID: -1, devID: 5, objGrpID: -1, objID: 2, pluginID: -1 }
      # - { devGrpID: -1, devID: 5, objGrpID: -1, objID: 3, pluginID: -1 }
      # - { devGrpID: -1, devID: 5, objGrpID: -1, objID: 4, pluginID: -1 }

exclude-filters:
  - name: blocklist1
    # Specify your filters as different elements in this array
    # by specifying an ID that you would like to be included.
    # A value of -1 is interpreted as any ID.

    # Each column in a given filter is combined with logical AND.
    # Multiple filters are combined with logical OR.
    rules:
      # Example: Exclude everything
      - { devGrpID: -1, devID: -1, objGrpID: -1, objID: -1, pluginID: -1 }
  - name: blocklist2
    rules:
      # Example: Exclude only devices 5 and 6
      - { devGrpID: -1, devID: 5, objGrpID: -1, objID: -1, pluginID: -1 }
      - { devGrpID: -1, devID: 6, objGrpID: -1, objID: -1, pluginID: -1 }

# Configure the settings for Schema Registry server if needed
schema-registry:
  # url: "http://SCHEMA_REGISTRY_SERVER_HOST:SCHEMA_REGISTRY_SERVER_PORT"

  # Subject for the schemas
  subject: sevone-data-publisher

status:
  # periodically (in seconds) print stats in the log if it's set to a positive integer. It's disabled by default.
  metricsLogInterval: 300
  http:
    # Configure the status HTTP page created by SDP
    # hosted at http://hostname:port/status
    enabled: true
    port: 8082
  https:
    # Configure the status HTTPS page created by SDP
    # hosted at https://hostname:secure_port/status
    enabled: false
    secure_port: 8443
    server_cert: /secrets/sdp/server.crt
    server_key: /secrets/sdp/server.key
    # private_key_password is an optional field
    private_key_password: password

# Output configuration
output:
  # Default settings for publishers, which can be overwritten by each publisher
  default:
    # Customize the message key format if needed. Available fields include:
    #     deviceId, deviceName, deviceIp, peerId, objectId, objectName,
    #     objectDesc, pluginId, pluginName, indicatorId, indicatorName,
    #     format, value, time, clusterName, peerIp
    # Default format is "deviceId:objectId".
    key-fields:
      - deviceId
      - objectId
    key-delimiter: ":"
    # Default to be kafka
    # Allowed values are: kafka, pulsar
    type: kafka
    kafka-producer:
      acks: -1
      retries: 0
      linger.ms: 10
      batch.size: 1000000
      request.timeout.ms: 60000
      max.in.flight.requests.per.connection: 2
    pulsar-producer:
      batchingMaxMessages: 1000
      blockIfQueueFull: true
      sendTimeoutMs: 30000
  publishers:
    - name: default-producer
      type: kafka
      topic: sdp
      isLive: true
      # version: 0.10.0.0
      # Kafka producer configuration options.
      # See https://kafka.apache.org/documentation, section 3.3 Producer Configs
      producer:
        # If bootstrap.servers is not defined, SDP will look for the bootstrap.servers
        # defined in output.default.kafka-producer.
        # Example: <your-kafka-ip>:<port>
        bootstrap.servers: null
        ## SSL setup
        # security.protocol: SSL
        ## SSL Server Authentication
        # ssl.ca.cert.location: server.crt
        ## SSL Client Authentication
        # ssl.client.cert.location: client.crt
        # ssl.client.key.location: client.key
        # ssl.client.key.password: password
        ## SASL configuration
        # sasl.mechanism: GSSAPI
        # sasl.kerberos.service.name: kafka
        # sasl.username: username
        # sasl.password: password
        # sasl.gssapi.useKeyTab: true
        # sasl.gssapi.storeKey: true
        # sasl.gssapi.keyTab: /secrets/sdp/sdp.keytab
        # sasl.gssapi.principal: sdp
        # sasl.gssapi.realm: EXAMPLE.COM
        # sasl.gssapi.kerberosconfpath: /config/kerberos/krb5.conf
        # sasl.gssapi.disablepafxfast: false
      filters:
        - default
    # Pulsar producer configuration options.
    - name: default-pulsar-producer
      type: pulsar
      topic: sdp-pulsar
      tenant: public
      namespace: default
      topic-type: persistent
      isLive: true
      client:
        # Example: pulsar://<your-pulsar-ip>:<port>
        serviceUrl: null
        connectionTimeoutMs: 10000            # Milliseconds
        # useTls: true
        # tlsTrustCertsFilePath: /path/to/server.crt
        # tlsAllowInsecureConnection: false
        # authPluginClassName: org.apache.pulsar.client.impl.auth.AuthenticationTls
        # authParams: tlsCertFile:client.crt,tlsKeyFile:client.key
        # operationTimeoutMs: 30000           # Milliseconds
        # numIoThreads: 1
        # tlsHostnameVerificationEnable: false
        # keepAliveIntervalSeconds: 30          # Seconds
      producer:
        compressionType: ZLIB
        # batchingMaxPublishDelayMicros: 1000    # Microseconds
        # chunkingEnabled: false
      filters:
        - default