• Overview
• Status Data Model
• How to publish status and dataset details events
  • Publish messages directly in message queue
  • Publish messages using endpoints
• Sample of status and dataset details message contracts
• Supported Stages and Statuses
• Restricting Field Size

Global Status monitoring is a mechanism that tracks the status of data journey/dataflows on the data platform. This mechanism helps to track the status of files, data, and records ingested through the File Service, Storage API, and specific DOMS until it is consumed by dependent services.

Every stage publishes status events to the message queue. Some stages, like WKS_SYNC, INGESTOR_SYNC, STORAGE_SYNC. etc., publish two statuses per record. The first status is for “In Progress” messages, and the second status is for a “SUCCESS” or “FAILED” message. From there on, the Status Collector picks up all status events and normalizes them to store in persistent storage for future reference. Then the Status Processor provides an API to query and check the status of past datasets.

The core services of Global Status Monitoring are the Status Collector and Status Processor.

Status Processor Service provides APIs that allow users to monitor the status of files, data, and records ingested through the File Service, Storage API, and specific DOMS until it is consumed by dependent services. The status indicates whether the dataflow has finished or not and if it is Successful or Failed.

For more detailed information, follow Status Processor tutorial

The Status Publisher Service provide APIs for publishing dataSet details and status messages to statuschangedtopic.

For more detailed information, follow Status Publisher tutorial

Data Model properties help users search for status with multiple or specific properties. Every request is tracked through a specific dataSetId or its associated correlationId. correlationId is a unique id that tracks the status of a request at various stages of the Data Platform.

The Status Data Model has multiple tables for tracking the stages of dataflow.

• DataSet Details - Dataset pertains to any data, such as file, collection of files, etc.
• Status - Holds the status of a dataflow.

There are two ways to publish status and dataset details in a message queue.

To publish status and dataset details directly to a message queue for a JAVA application, follow the steps below:

1. Add the os core common library as a dependency - There are models, classes, and interfaces defined in the core common lib from Azure.
   Ensure that you select the correct version of the library which includes the StatusDetails and DatasetDetails classes, 0.13.0 or higher.

   Core Common Library contents for GSM

   • Models - StatusDetails and DatasetDetails - Use these two models to publish status and dataset details.
   • Utility - AttributesBuilder - Creates an attributes map which is a required method of IEventPublisher to publish status or dataset details.
     The attributes map consists of the slb partition id and correlation id.
   • Publisher Interface - IEventPublisher - This is the interface that the cloud provider must implement to produce status and dataset details.
     It contains a method which accepts the Message array and Attributes map. Message is an interface that is implemented by both Status and Dataset Details.

  <dependency>
    <groupId>org.opengroup.osdu</groupId>
    <artifactId>os-core-common</artifactId>
    <version>0.13.0</version>
  </dependency>

2. Add core-lib-azure library as dependency - provides a publisher facade to publish messages to the ServiceBus, version 0.13.0 or higher.

  <dependency>
    <groupId>org.opengroup.osdu</groupId>
    <artifactId>core-lib-azure</artifactId>
    <version>0.13.0</version>
  </dependency>

3. Publisher to identify all scenarios to publish Status/Dataset Details - You should determine all possible scenarios in which either Status or Dataset Details need to be published.
   A service can publish multiple sets of both Status and Dataset Details.
4. Cloud Implementation to publish Status/Dataset Details
   • Provide an implementation of the IEventPublisher interface from the os core common library. The publish method of IEventPublisher must be implemented to publish an event in statuschangedtopic that accepts:
     • Array of type Message: Message is an interface implemented by both StatusDetails and DatasetDetails classes. The Message array can be either Status or Dataset Details. For an example , refer to section Sample of status and dataset details message
     • Map of key-value pair attributes: Key and value are of type string. Examples: "correlation-id": "1235345" and "slb-partition-id": "user defined value".
5. Create PublisherInfo to provide needed information for publishing:
   • batch - An array of messages, such as Status or DataSet Details.
   • serviceBusTopicName - The name of the ServiceBus topic that is statuschangedtopic.
6. Call method publishMessage from MessagePublisher class that accepts:
   • DpsHeaders - Must contain "correlation-id" and "slb-partition-id".
   • PublisherInfo - The ServiceBus topic name and data to publish.
7. Enable ServiceBus publishing by providing the following configuration - azure.serviceBus.enabled = true

  public class StatusEventPublisher implements IEventPublisher {

  private final MessagePublisher messagePublisher;
  private final ServiceBusConfig serviceBusConfig;
  private final DpsHeaders dpsHeaders;
  private final JaxRsDpsLog log;

  @Override
  public void publish(Message[] messages, Map<String, String> attributesMap) throws CoreException {
      PublisherInfo publisherInfo = PublisherInfo.builder()
              .batch(messages)
              .serviceBusTopicName(serviceBusConfig.getServiceBusTopic())
              .build();
      messagePublisher.publishMessage(dpsHeaders, publisherInfo);
      log.info("Status event generated successfully");
  }

  public class StatusPublisher {
      private static final String FAILED_TO_PUBLISH_STATUS = "Failed to publish status ";
      private static final String PUBLISH_STATUS_STARTED = "Publish Status started";
   
      private final IEventPublisher statusEventPublisher;
      private final AttributesBuilder attributesBuilder;
      private final JaxRsDpsLog log;
   
      private void publish(List<StatusDetails> statusDetailsList) {
          try {
              log.info(PUBLISH_STATUS_STARTED);
              Map<String, String> attributesMap = attributesBuilder.createAttributesMap();
               
              statusEventPublisher.publish(statusDetailsList.toArray(new StatusDetails[0]), attributesMap);
          } catch (CoreException e) {
              log.warning(FAILED_TO_PUBLISH_STATUS + e.getMessage());
              throw new ApplicationException(FAILED_TO_PUBLISH_STATUS, e);
          }
      }
  }

Note: We have an Azure implementation of Global Status Monitoring. Services that are not part of the OSDU AKS cluster have to use /status and /datasetDetails endpoints of the Status Processor service. The Status Processor service publishs status and dataset details in statuschangedtopic.

There are two endpoints to publish status messages in the queue. These messages can be used by applications outside the OSDU cluster if they want to use GSM to track the status.

• Publish Status - POST /status endpoint
• Publish Dataset Details - POST /dataset-details endpoint

Follow Status Publisher tutorial for more information related to publish endpoints.

There are two kinds of status messages, DataSet Details and Status, but they are published into the same Azure ServiceBus statuschangedtopic.

• DataSet Details

{
  "message":{
    "data":[
      {
        "kind": "datasetDetails",
        "properties": {
          "correlationId": "1cc462e9-9a39-48b9-84c3-9d53b18b8089",
          "datasetId": "opendes:dataset--File.Generic:332b55d5-0cf2-4474-a3e5-d3cff0c5746a",
          "datasetVersionId": "opendes:dataset--File.Generic:332b55d5-0cf2-4474-a3e5-d3cff0c5746a:23423432",
          "datasetType": "FILE",
          "recordCount": 1,
          "timestamp": 1625221800
        }
      }
    ],
    "account-id":"opendes",
    "slb-partition-id":"opendes",
    "correlation-id":"3f41d31b-862f-40ce-9749-ee26a3f714f2"
  }
}

• Status

{
  "message":{
    "data":[
      {
        "kind": "status",
        "properties": {
          "correlationId": "1cc462e9-9a39-48b9-84c3-9d53b18b8089",
          "recordId": "opendes:wellbore:osdudemo-ATVMxMDEzTVMxMDQ",
          "recordIdVersion": "opendes:wellbore:osdudemo-ATVMxMDEzTVMxMDQ:23423432",
          "stage": "STORAGE_SYNC",
          "status": "FAILED",
          "message": "acl is not valid",
          "errorCode": 400,
          "userEmail": "test@email.com",
          "timestamp": 1625221800
        }
      }
    ],
    "account-id":"opendes",
    "slb-partition-id":"opendes",
    "correlation-id":"3f41d31b-862f-40ce-9749-ee26a3f714f2"
  }
}

The stage shows the current activity of the record. For example if the record is in WKS transformation, this will be at the WKS_SYNC stage.

The status of the record shows if it was processed as Successful or Failed.

• Status Fields Max Allowed Size