Monitor the Openflow Connector for MongoDB

The connector writes the metrics described on this page to your Openflow event table. A subset of these metrics is also visualized in the Openflow Connectors Dashboard. For generic Openflow telemetry, see Openflow telemetry.

Metrics

All metrics include the same resource and connector attributes as application metrics. Collection-level metrics also include collection attributes. Some metrics also include additional attributes.

MetricTypeResolutionDescription
db.table.statusGaugeCollectionCurrent replication status of the collection. Reported every 60 seconds for every collection in replication. See Replication status values.
db.table.status.countGaugeConnectorCount of collections in each status. Reported every 60 seconds, as long as any collections are in replication. See Replication status values.
db.last.ingestion.timeGaugeCollectionTime when data was last written to the destination table, in milliseconds since the UNIX epoch. Reported every 60 seconds for up to 3 days after the last write, even when no new data arrives.
db.records.ingestedCounterCollectionNumber of records written to Snowflake in each replication phase, indicated by the db.replication.phase attribute. In the SNAPSHOT phase, the value represents the number of records written to the destination table. In the INCREMENTAL phase, the value represents the number of records written to journal tables. Reported every 60 seconds for every collection that received data since the last report.
db.commit.to.merge.latencyGaugeCollectionHighest time, in milliseconds, from when a change was committed in the source database until it was merged into the destination table. Reported every 60 seconds for every collection that received data since the last report.
db.commit.to.capture.latencyGaugeConnectorHighest time, in milliseconds, from when a change was committed in the source database until the connector read it. Reported every 60 seconds while the connector receives data.
db.capture.to.journal.latencyGaugeConnectorHighest time, in milliseconds, from when the connector read a change until it was written to the journal table in Snowflake. Reported every 60 seconds while the connector receives data.

Collection-level metrics support up to 75,000 collections in a single runtime. Above that limit, metrics might be lost.

Source object attributes

Collection-level metrics identify the source collection with the following record attributes:

AttributeDescription
source.schema.nameName of the source database.
source.table.nameName of the source collection.

Replication status values

The db.table.status metric encodes the replication status of a collection as a number. The mapping is as follows:

ValueNameDescription
0-The collection is no longer replicated, for example because it was removed from replication. Reported once.
1NEWThe collection is scheduled for replication, but replication hasn’t started.
2SNAPSHOT_REPLICATIONThe connector is copying the existing documents of the collection.
3INCREMENTAL_REPLICATIONThe connector is replicating changes.
4FAILEDReplication has permanently stopped for an unknown reason.
4002CONTENT_MERGE_FAILEDThe connector failed to merge FlowFile content.
4005MERGE_QUERY_FAILEDThe merge query in Snowflake failed after the maximum number of retries.
4006MERGE_QUERY_UNKNOWN_STATUSThe merge query in Snowflake returned an unknown status.
4009MISSING_REQUIRED_ATTRIBUTESA FlowFile is missing required attributes, or has invalid values for them.
4011NO_COLUMNS_IN_TABLEThe schema generated for the collection has no columns.
4015REMOVED_FROM_REPLICATIONThe collection was removed from the source database, no longer matches the collection filter, or is in a permanently failed state.
4016SCHEMA_CONVERSION_ERRORThe collection schema couldn’t be converted to the expected format.
4018SNOWFLAKE_OBJECT_ALREADY_EXISTSThe Snowflake object already exists and can't be created.
4019SNOWFLAKE_OBJECT_OPERATION_FAILEDThe connector failed to create, update, or manage a Snowflake object.
4022FLOWFILE_READ_ERRORFlowFile content couldn’t be read or parsed.
4024SNOWPIPE2_UPLOAD_FAILEDSnowflake rejected data sent with Snowpipe Streaming because of invalid record content.

Values of 4001 and higher mean that replication has permanently stopped for the listed reason.

Alerts

The following alert templates in the OPENFLOW template group are specific to this connector:

TemplateDescriptionSettings
Table replication failure alert

Fires for each collection whose replication failed during the lookback window. Can be configured for the account, a deployment, a runtime, or a connector.

  • Lookback window (hours):
    Default: 1.
  • Table names filter: Optional. A comma-separated list of the collections to monitor. Leave empty to monitor all collections.
Openflow data staleness alert

Fires for each of the specified collections when no new data was written to its destination table for longer than the staleness threshold. Can be configured for the account, a deployment, a runtime, or a connector.

  • Table names filter: Required. A comma-separated list of the collections to monitor.
  • Staleness threshold (hours):
    Default: 4.
  • Lookback window (hours):
    Default: 6.

For more information and generic Openflow alerts, see Set up alerts for Openflow.

Example queries

The following queries assume that Openflow sends telemetry to the default event table, SNOWFLAKE.TELEMETRY.EVENTS. If you use a different event table, substitute its name.

Find replication failures

This query returns the collections whose latest status in the last hour is failed:

SELECT
  record_attributes:"connector.name"::STRING AS connector,
  record_attributes:"source.schema.name"::STRING AS source_schema,
  record_attributes:"source.table.name"::STRING AS source_table,
  value::NUMBER AS status,
  timestamp
FROM SNOWFLAKE.TELEMETRY.EVENTS
WHERE timestamp > DATEADD(hour, -1, SYSDATE())
  AND record_type = 'METRIC'
  AND record:metric:name = 'db.table.status'
QUALIFY ROW_NUMBER() OVER (
    PARTITION BY connector, source_schema, source_table
    ORDER BY timestamp DESC) = 1
  AND status >= 4;

Find destination tables without recent data

This query returns the collections that had no data written to the destination table in the last hour. A collection without changes in the source database also appears in the results.

SELECT
  record_attributes:"connector.name"::STRING AS connector,
  record_attributes:"source.schema.name"::STRING AS source_schema,
  record_attributes:"source.table.name"::STRING AS source_table,
  TO_TIMESTAMP_LTZ(MAX(value::NUMBER), 3) AS last_ingestion_time
FROM SNOWFLAKE.TELEMETRY.EVENTS
WHERE timestamp > DATEADD(hour, -2, SYSDATE())
  AND record_type = 'METRIC'
  AND record:metric:name = 'db.last.ingestion.time'
GROUP BY connector, source_schema, source_table
HAVING last_ingestion_time < DATEADD(hour, -1, CURRENT_TIMESTAMP());

Plot replication latency over time

This query returns the latency from source commit to destination table for each collection over the last 24 hours. To plot it, show the results as a line chart with timestamp on the X-axis and latency_seconds on the Y-axis.

SELECT
  timestamp,
  record_attributes:"connector.name"::STRING AS connector,
  record_attributes:"source.schema.name"::STRING AS source_schema,
  record_attributes:"source.table.name"::STRING AS source_table,
  value::NUMBER / 1000 AS latency_seconds
FROM SNOWFLAKE.TELEMETRY.EVENTS
WHERE timestamp > DATEADD(day, -1, SYSDATE())
  AND record_type = 'METRIC'
  AND record:metric:name = 'db.commit.to.merge.latency'
ORDER BY timestamp;

Count records written per replication phase

This query returns the number of records written for each collection per hour and replication phase in the last 24 hours:

SELECT
  DATE_TRUNC(hour, timestamp) AS hour,
  record_attributes:"connector.name"::STRING AS connector,
  record_attributes:"source.schema.name"::STRING AS source_schema,
  record_attributes:"source.table.name"::STRING AS source_table,
  record_attributes:"db.replication.phase"::STRING AS phase,
  SUM(value::NUMBER) AS records
FROM SNOWFLAKE.TELEMETRY.EVENTS
WHERE timestamp > DATEADD(day, -1, SYSDATE())
  AND record_type = 'METRIC'
  AND record:metric:name = 'db.records.ingested'
GROUP BY hour, connector, source_schema, source_table, phase
ORDER BY hour, connector, source_schema, source_table;