← Cloudflare Pipelines / pipelines / observability

Метрики и аналитика

Pipelines предоставляет метрики, которые позволяют измерять объём принятых, обработанных и доставленных в приёмники данных.

Метрики, отображаемые в Панель управления Cloudflare ↗ запрашиваются из Cloudflare GraphQL Analytics API. Вы можете получить доступ к метрикам программно через GraphQL или HTTP-клиент.

Pipelines экспортирует указанные ниже метрики в рамках pipelinesOperatorAdaptiveGroups набор данных. Эти метрики отслеживают данные, считанные и обработанные операторами конвейера.

Подробный разбор причин, по которым события были отклонены (включая конкретные типы ошибок, такие как missing_field, type_mismatch, parse_failure, а также null_value), см. Метрики пользовательских ошибок.

pipelinesOperatorAdaptiveGroups набор данных предоставляет следующие измерения для фильтрации и группировки запросов:

• pipelineId - ID конвейера
• streamId - ID исходного потока
• datetime - временная метка операции
• date - временная метка операции, округленная до начала дня
• datetimeHour - временная метка операции, округленная до начала часа

Pipelines экспортирует указанные ниже метрики в рамках pipelinesSinkAdaptiveGroups набор данных. Эти метрики отслеживают доставку данных в приёмники.

pipelinesSinkAdaptiveGroups набор данных предоставляет следующие измерения для фильтрации и группировки запросов:

• pipelineId - ID конвейера
• sinkId - ID целевого приёмника
• datetime - временная метка операции
• date - временная метка операции, округленная до начала дня
• datetimeHour - временная метка операции, округленная до начала часа

Pipelines отслеживает события, которые отбрасываются при обработке из-за ошибок десериализации. Если структурированный поток получает события, не соответствующие заданной схеме, такие события принимаются при приёме, но отбрасываются при обработке. Показатель pipelinesUserErrorsAdaptiveGroups предоставляет информацию об этих отброшенных событиях, показывая, какие события были отброшены и почему. Полную схему этого набора данных можно изучить с помощью GraphQL интроспекция.

pipelinesUserErrorsAdaptiveGroups набор данных предоставляет следующие измерения для фильтрации и группировки запросов:

• pipelineId - ID конвейера
• errorFamily - категория ошибки (например, deserialization)
• errorType - конкретный тип ошибки в рамках семейства
• date - дата ошибки, округленная до начала дня
• datetime - временная метка ошибки
• datetimeHour - временная метка ошибки, округленная до начала часа
• datetimeMinute - временная метка ошибки, округленная до начала минуты

Аналитика по каждому конвейеру доступна на панели управления Cloudflare. Чтобы посмотреть текущие и исторические метрики конвейера, выполните следующее:

1. Войдите в Панель управления Cloudflare ↗ и выберите свой аккаунт.
2. Перейдите в Pipelines > Pipelines.
3. Выберите конвейер.
4. Перейдите в Метрики вкладку для просмотра метрик или Ошибки вкладку, чтобы просмотреть отброшенные события.

При необходимости можно выбрать временной интервал для запроса. По умолчанию используются последние 24 часа.

Вы можете программно запрашивать аналитику по своим конвейерам через GraphQL Analytics API. Этот API запрашивает те же наборы данных, что и Cloudflare dashboard, и поддерживает GraphQL интроспекция.

Для наборов данных Pipelines GraphQL требуется accountTag фильтр с вашим Cloudflare Account ID.

Этот запрос возвращает общее количество байт и записей, прочитанных конвейером из потоков, а также ошибки декодирования, если они есть.

query PipelineOperatorMetrics(
	$accountTag: String!
	$pipelineId: String!
	$datetimeStart: Time!
	$datetimeEnd: Time!
) {
	viewer {
		accounts(filter: { accountTag: $accountTag }) {
			pipelinesOperatorAdaptiveGroups(
				limit: 10000
				filter: {
					pipelineId: $pipelineId
					streamId_neq: ""
					datetime_geq: $datetimeStart
					datetime_leq: $datetimeEnd
				}
			) {
				sum {
					bytesIn
					recordsIn
					decodeErrors
				}
			}
		}
	}
}

Этот запрос возвращает подробные метрики данных, записанных в конкретный приёмник, включая статистику по файлам и сжатию.

query PipelineSinkMetrics(
	$accountTag: String!
	$pipelineId: String!
	$sinkId: String!
	$datetimeStart: Time!
	$datetimeEnd: Time!
) {
	viewer {
		accounts(filter: { accountTag: $accountTag }) {
			pipelinesSinkAdaptiveGroups(
				limit: 10000
				filter: {
					pipelineId: $pipelineId
					sinkId: $sinkId
					datetime_geq: $datetimeStart
					datetime_leq: $datetimeEnd
				}
			) {
				sum {
					bytesWritten
					recordsWritten
					filesWritten
					rowGroupsWritten
					uncompressedBytesWritten
				}
			}
		}
	}
}

Этот запрос возвращает сводку событий, отклонённых из-за ошибок проверки схемы, сгруппированную по типу ошибки и отсортированную по частоте.

query GetPipelineUserErrors(
	$accountTag: String!
	$pipelineId: String!
	$datetimeStart: Time!
	$datetimeEnd: Time!
) {
	viewer {
		accounts(filter: { accountTag: $accountTag }) {
			pipelinesUserErrorsAdaptiveGroups(
				limit: 100
				filter: {
					pipelineId: $pipelineId
					datetime_geq: $datetimeStart
					datetime_leq: $datetimeEnd
				}
				orderBy: [count_DESC]
			) {
				count
				dimensions {
					date
					errorFamily
					errorType
				}
			}
		}
	}
}

Пример ответа:

{
	"data": {
		"viewer": {
			"accounts": [
				{
					"pipelinesUserErrorsAdaptiveGroups": [
						{
							"count": 679,
							"dimensions": {
								"date": "2026-02-19",
								"errorFamily": "deserialization",
								"errorType": "missing_field"
							}
						},
						{
							"count": 392,
							"dimensions": {
								"date": "2026-02-19",
								"errorFamily": "deserialization",
								"errorType": "type_mismatch"
							}
						},
						{
							"count": 363,
							"dimensions": {
								"date": "2026-02-19",
								"errorFamily": "deserialization",
								"errorType": "parse_failure"
							}
						},
						{
							"count": 44,
							"dimensions": {
								"date": "2026-02-19",
								"errorFamily": "deserialization",
								"errorType": "null_value"
							}
						}
					]
				}
			]
		}
	},
	"errors": null
}

Чтобы отфильтровать по конкретному типу ошибки, добавьте errorType в фильтр:

pipelinesUserErrorsAdaptiveGroups(
	limit: 100
	filter: {
		pipelineId: $pipelineId
		datetime_geq: $datetimeStart
		datetime_leq: $datetimeEnd
		errorType: "type_mismatch"
	}
	orderBy: [count_DESC]
)

Чтобы получить ошибки по всем конвейерам в аккаунте, не указывайте pipelineId фильтр и включите pipelineId в измерениях:

pipelinesUserErrorsAdaptiveGroups(
	limit: 100
	filter: {
		datetime_geq: $datetimeStart
		datetime_leq: $datetimeEnd
	}
	orderBy: [count_DESC]
) {
	count
	dimensions {
		pipelineId
		errorFamily
		errorType
	}
}