Integrating Prometheus with the DolphinDB Rule Engine to Build an Agile Monitoring Solution

Base path: /api/v1

Status codes:

• Returns 2xx when a request is executed successfully.

• Returns 400 Bad Request when parameters are missing or incorrect.

• Returns 422 Unprocessable Entity when an expression cannot be executed.

• Returns 503 Service Unavailable when a query times out or is aborted.

Response format: JSON. The response envelope format is as follows:

{
  "status": "success" | "error",
  "data": <data>,
  // Only set if status is "error". The data field may still hold
  // additional data.
  "errorType": "<string>",
  "error": "<string>",
  // Only set if there were warnings while executing the request.
  // There will still be data in the data field.
  "warnings": ["<string>"],
  // Only set if there were info-level annotations while executing the request.
  "infos": ["<string>"]
}

The following endpoint evaluates an expression at a single point in time.

GET /api/v1/query
POST /api/v1/query

URL query parameters:

• query=<string>: Prometheus expression query string.

• time=<rfc3339 | unix_timestamp>: Optional. Evaluation timestamp. If omitted, the current server time is used.

• timeout=<duration>: Optional. Evaluation timeout.

• limit=<number>: Optional. Maximum number of series to return.

Time Parameter Handling

Prometheus stores all time-series data in UTC time , while most client execution environments use local time as their timezone, so a time conversion is required when constructing query requests. The specific conversion method depends on the client tool you are using:

• curl command: You need to manually convert local time to UTC time at the command line. For example, convert local time "2025-11-10 16:14:27" to UTC time "2025-11-10T08:14:27Z" before passing it in.

• DolphinDB HTTP Client: DolphinDB uses local time (e.g., UTC+8) by default for data processing and queries. When querying a metric at a specific point in time, you can use the built-in time function gmtime to convert to UTC before passing it in. When retrieving data from Prometheus for alerting rules processing, you also need to convert it to local time using function localtime .

Example: Query the metric lastMinuteNetworkRecv at time 2025-11-10T08:14:27Z (UTC).

curl 'localhost:9090/api/v1/query?query=lastMinuteNetworkRecv&time=2025-11-10T08:14:27Z'

Result:

text: '{
  "status":"success",
  "data":{
    "resultType":"vector",
    "result":[
      {
        "metric":{
          "__name__":"lastMinuteNetworkRecv",
          "instance":"host:port",
          "job":"DolphinDB"
        },
        "value":[1762762467,"1952900"]
      }
    ]
  }
}'
elapsed: 0.001651
headers: 'HTTP/1.1 200 OK Content-Type: application/json Vary: Origin Date: Mon, 10 Nov 2025 08:51:03 GMT Content-Length: 193 '
responseCode: 200

Users can query a time range of data through the query_range endpoint. A single query for one time series can return a maximum of 11,000 data points.

GET /api/v1/query_range
POST /api/v1/query_range

URL query parameters:

• The query , limit , and timeout parameters are the same as those for Instant Query and are not repeated here.

• start=<rfc3339 | unix_timestamp>: Start timestamp, inclusive.

• end=<rfc3339 | unix_timestamp>: End timestamp, inclusive.

• step=<duration | float>: Query resolution step, in duration format or as a float number of seconds.

Example: Query the metric lastMinuteNetworkRecv over the time range 2025-11-10T08:14:00 to 2025-11-10T08:15:00 with a step of 15s.

curl 'localhost:9090/api/v1/query_range?query=lastMinuteNetworkRecv&start=2025-11-10T08:14:00Z&end=2025-11-10T08:15:00Z&step=15s'

Return:

text: '{
  "status":"success",
  "data":{
    "resultType":"matrix",
    "result":[
      {
        "metric":{
          "__name__":"lastMinuteNetworkRecv",
          "instance":"host:ip",
          "job":"DolphinDB"
        },
        "values":[
          [1762762440,"1985379"],
          [1762762455,"1952832"],
          [1762762470,"1952900"]
        ]
      }
    ]
  }
}'
elapsed: 0.001218
headers: 'HTTP/1.1 200 OK Content-Type: application/json Vary: Origin Date: Mon, 10 Nov 2025 09:03:54 GMT Content-Length: 242 '
responseCode: 200

// Define the table schema.
colName = `metricName`timestamp`value`instance`job;
colType = [SYMBOL, DATETIME, DOUBLE, STRING, STRING];
// Create and persist the dataReceiveTable Stream Table to receive real-time fetched data.
enableTableShareAndPersistence(
    table=streamTable(1000:0, colName, colType),
    tableName=`dataReceiveTable,
    cacheSize=1000000
);
// Create a DFS table to store data processed by the Rule Engine.
create table "dfs://metricsDB"."warn"(
    metric_name SYMBOL,
    timestamp TIMESTAMP,
    value DOUBLE,
    instance SYMBOL,
    job SYMBOL,
    rule BOOL[]
)
partitioned by timestamp
sortColumns = `metric_name`timestamp

Here, server hardware resource data is obtained by scheduled polling. Two functions are defined for this purpose: query and pullData . The polling approach minimizes intrusiveness to the monitored target as much as possible, which better matches the requirement for lightweight monitoring.

• query: Sends an instant query request and parses the HTTP response into table data.

• pullData: Calls query to poll the Prometheus server periodically and obtain data.

The parameter metrics is a string vector containing the names of the hardware resource metrics to be queried, and interva l is the polling interval in seconds. It is recommended to keep it consistent with the scrape_interval setting in the Prometheus configuration file.

loadPlugin("httpClient");
def dict2Table(dataDict){
    v = dataDict.value
    metricDict=dataDict.metric
    ts =  timestamp(long(v[0]*1000))
    val = v[1].double()
    return table(
        take(metricDict["__name__"], 1) as metricName, 
        localtime(ts) as timestamp,
        val as value,
        take(metricDict["instance"], 1) as instance, 
        take(metricDict["job"], 1) as job
    )
}
def query(host, port, metrics){
    url = host + ":" +  port + '/api/v1/query'
    params = {
        query: metrics
    }
    res = httpClient::httpGet(url, params)
    rst = parseExpr(res.text).eval()
    allMetrics = select instance, timestamp, metricName, value, job
        from each(dict2Table, rst.data.result).unionAll() 
        order by instance, metricName
    return allMetrics
}
// Define the real-time data pulling function. The interval can match the Prometheus scrape interval.
def pullData(host, port, metrics, table, interval=15){
    do{
        data = query(host, port, metrics)
        tableInsert(table, data)
        sleep(interval*1000)
    }while(true)
}
// Replace the following values based on your monitoring requirements.
host = "192.198.1.36"
port = 9090
metrics = '{__name__=~"cpuUsage|diskFreeSpaceRatio|networkRecvRate"}'
targetTable = objByName("metricsStream") 
jobId = stringFormat("pullPromethus_%W_%i", host, port).strReplace(".", "_")
submitJob(jobId, "get Promethus metrics", pullData, host, port, metrics, targetTable)

The DolphinDB Rule Engine is a concise, efficient, and extensible module for rule checking and alerting. It supports centralized O&M of monitoring rules, including adding, modifying, and deleting rules, which can greatly improve operational efficiency.

The DolphinDB Rule Engine marks inserted data according to user-defined rule sets, outputs the marked data to a specified table, and invokes a callback function so that users can process marked data according to custom logic. For details, refer to creatRuleEngine .

In this example, ruleSets is initialized as follows:

// Initialize ruleSets.
index = array(SYMBOL, 0)
index.append!(`cpuUsage`diskFreeSpaceRatio`networkRecvRate NULL)
rule = [[<value > 90>], [<value < 0.1>], [<value > 10000000>], [<value == NULL >]];
ruleSets = dict(index, rule);

The callback function can be used to implement data cleansing, data writing, alerting, and other logic. In this example, one alert delivery method is provided: email alerts. Users can integrate their existing enterprise alerting system with DolphinDB message/protocol plugins based on actual application scenarios.

Use the DolphinDB httpClient plugin to send alert emails.

def send2Email(result){
    if(result.rule[0]){
        text = stringFormat("Server %W service/resource %W %W alert. Current value: %.2f", 
            result.instance[0], result.job[0], result.metricName[0], result.value[0])
        res = httpClient::sendEmail('MailFrom@xxx.com','password','MailDestination@xxx.com','Data Alert', text);
    }
}

Finally, consume data by subscribing to the stream table with the subscribeTable function.

// Create the Rule Engine.
colName = `metricName`timestamp`value`instance`job;
colType = [SYMBOL, DATETIME, DOUBLE, STRING, STRING];
dummy = table(1:0, colName, colType)
ruleEngine = createRuleEngine(
        name="ruleEngine",
        ruleSets=ruleSets,
        dummyTable=dummy,
        outputColumns=`metricName`timestamp`value`instance`job,
        outputTable=loadTable("dfs://metricsDB", "warn"),
        policy="all",
        ruleSetColumn="metricName",
        callback=send2Wechat
)
subscribeTable(
    tableName="dataReceiveTable", 
    actionName="write2ruleEngine", 
    handler=getStreamEngine("ruleEngine"), 
    msgAsTable=true,
    batchSize=10000,
    throttle=3
)

Note: This tutorial assumes that alert events occur at a per-second frequency, which is suitable for most application scenarios. If the actual alert density is very high, for example, thousands of alerts per second, it is not recommended to call I/O-related functions directly in the callback because the callback is executed synchronously. In this case, you can write alert results to a stream table through outputTable , then subscribe to that stream table and implement the alert delivery logic in the handler.

After initializing the engine, run getStreamEngineStat() to view the engine state.

The engine output ( outputTable ) contains the original monitoring information as well as the rule evaluation result rule . rule is a boolean array. You can filter data that exceeds the threshold with rule[0] = true .

select * from loadTable("dfs://metricsDB", "warn") 
where rule[0] = true
order by timestamp desc limit 100

As monitoring conditions change, for example, due to scaling hardware resources for containers or changes in business load, monitoring thresholds often need to be adjusted dynamically. The DolphinDB rule engine supports online dynamic rule adjustment. Rules can be added, modified, or deleted at runtime through the updateRule and deleteRule functions.

Example of updating a monitoring rule

Call updateRule to update the CPU threshold online to 85 percent.

updateRule("ruleEngine","cpuUsage",[<value>85>])

After the update, insert a new record with a new timestamp.

ruleEngine.append!(table("cpuUsage" as metric_name, datetime(2025.12.01 10:00:00) as timestamp, 88.0 as value, "localhost:8848" as instance, "DolphinDB" as job))
getRules(["ruleEngine"])

After the modification, the rule engine successfully outputs the new alert information.