Мониторинг водяной мельницы

Мониторинг Watermill в реальном времени с использованием Prometheus

Метрики

Watermill можно мониторить, используя декораторы для издателей/подписчиков и промежуточное ПО для обработчиков. Мы предоставляем реализацию по умолчанию с использованием официального клиента Prometheus для Go.

Пакет components/metrics экспортирует PrometheusMetricsBuilder, который предоставляет удобные функции для оборачивания издателей, подписчиков и обработчиков для обновления соответствующего реестра Prometheus:

Полный исходный код: github.com/ThreeDotsLabs/watermill/components/metrics/builder.go

// ...
// PrometheusMetricsBuilder предоставляет методы для декорирования издателей, подписчиков и обработчиков.
type PrometheusMetricsBuilder struct {
    // PrometheusRegistry может заполнять существующий реестр Prometheus или быть пустым для использования реестра по умолчанию.
    PrometheusRegistry prometheus.Registerer

    Namespace string
    Subsystem string
}

// AddPrometheusRouterMetrics - это удобная функция для добавления промежуточного ПО метрик для всех обработчиков на маршрутизаторе сообщений. Он также декорирует издателей и подписчиков обработчиков.
func (b PrometheusMetricsBuilder) AddPrometheusRouterMetrics(r *message.Router) {
// ...

Обертывание издателей, подписчиков и обработчиков

Если вы используете маршрутизатор Watermill (что рекомендуется в большинстве случаев), можно использовать удобную функцию AddPrometheusRouterMetrics, чтобы убедиться, что все обработчики, добавленные к этому маршрутизатору, обернуты для обновления реестра Prometheus, а также их издатели и подписчики:

Полный исходный код: github.com/ThreeDotsLabs/watermill/components/metrics/builder.go

// ...
// AddPrometheusRouterMetrics - это удобная функция для добавления промежуточного ПО метрик для всех обработчиков на маршрутизаторе сообщений. Он также декорирует издателей и подписчиков обработчиков.
func (b PrometheusMetricsBuilder) AddPrometheusRouterMetrics(r *message.Router) {
    r.AddPublisherDecorators(b.DecoratePublisher)
    r.AddSubscriberDecorators(b.DecorateSubscriber)
    r.AddMiddleware(b.NewRouterMiddleware().Middleware)
}
// ...

Пример использования AddPrometheusRouterMetrics:

Полный исходный код: github.com/ThreeDotsLabs/watermill/_examples/basic/4-metrics/main.go

// ...
    // Мы оставляем параметры `namespace` и `subsystem` пустыми
    metricsBuilder := metrics.NewPrometheusMetricsBuilder(prometheusRegistry, "", "")
    metricsBuilder.AddPrometheusRouterMetrics(router)
// ...

В приведенном выше фрагменте кода мы оставляем параметры namespace и subsystem пустыми. Библиотека клиента Prometheus использует эти параметры для префикса имен метрик. Вы можете использовать namespace или subsystem, но обратите внимание, что это повлияет на имена метрик, поэтому вам нужно будет настроить панель мониторинга Grafana соответствующим образом.

Независимые издатели и подписчики также могут быть декорированы с использованием специальных методов PrometheusMetricsBuilder:

Полный исходный код: github.com/ThreeDotsLabs/watermill/_examples/basic/4-metrics/main.go

// ...
    subWithMetrics, err := metricsBuilder.DecorateSubscriber(pubSub)
    if err != nil {
        panic(err)
    }
    pubWithMetrics, err := metricsBuilder.DecoratePublisher(pub)
    if err != nil {
        panic(err)
    }
// ...

Открыть конечную точку /metrics

Согласно принципу работы Prometheus, сервис должен открывать HTTP-конечную точку для сбора данных. Обычно это GET-конечная точка, и путь обычно /metrics.

Для предоставления этой конечной точки существуют две удобные функции, одна использует ранее созданный Реестр Prometheus, а другая одновременно создает новый Реестр:

Полный исходный код: github.com/ThreeDotsLabs/watermill/components/metrics/http.go

// ...
// CreateRegistryAndServeHTTP устанавливает HTTP-сервер по указанному адресу для открытия конечной точки /metrics для Prometheus.
// Он возвращает новый Реестр Prometheus (для регистрации метрик) и функцию отмены для выключения сервера.
func CreateRegistryAndServeHTTP(addr string) (registry *prometheus.Registry, cancel func()) {
	registry = prometheus.NewRegistry()
	return registry, ServeHTTP(addr, registry)
}

// ServeHTTP устанавливает HTTP-сервер по указанному адресу для открытия конечной точки /metrics для Prometheus.
// Он принимает существующий Реестр Prometheus и возвращает функцию отмены для выключения сервера.
func ServeHTTP(addr string, registry *prometheus.Registry) (cancel func()) {
// ...

Вот пример использования:

Полный исходный код: github.com/ThreeDotsLabs/watermill/_examples/basic/4-metrics/main.go

// ...
	prometheusRegistry, closeMetricsServer := metrics.CreateRegistryAndServeHTTP(*metricsAddr)
	defer closeMetricsServer()

	// Мы оставляем пространство и подсистему пустыми
	metricsBuilder := metrics.NewPrometheusMetricsBuilder(prometheusRegistry, "", "")
	metricsBuilder.AddPrometheusRouterMetrics(router)
// ...

Пример приложения

Чтобы понять, как работает панель управления на практике, вы можете обратиться к примеру метрик.

Следуйте инструкциям в README примера, чтобы запустить его и добавить источник данных Prometheus в Grafana.

Панель управления Grafana

Мы подготовили панель управления Grafana для использования с упомянутой выше реализацией метрик. Она предоставляет базовую информацию о пропускной способности, уровне сбоев и длительности публикации/обработки.

Если вы хотите просмотреть эту панель управления локально, вы можете воспользоваться примером приложения.

Дополнительную информацию о метриках, экспортируемых в Prometheus, смотрите в разделе Экспортируемые метрики.

Импорт панели управления

Чтобы импортировать панель управления Grafana, выберите Панель/Управление в левом меню, затем нажмите +Импорт.

Введите URL панели управления https://grafana.com/dashboards/9777 (или только ID, 9777), затем нажмите Загрузить.

Затем выберите источник данных Prometheus, используемый для сбора данных с конечной точки /metrics. Нажмите Импорт и вы готовы!

Экспортируемые метрики

Ниже приведен список всех метрик, зарегистрированных в реестре Prometheus с помощью PrometheusMetricsBuilder.

Дополнительную информацию о типах метрик Prometheus можно найти в документации Prometheus.

Кроме того, каждая метрика имеет метку node, предоставленную Prometheus, соответствующую экземпляру источника метрики, и job, указанный в качестве названия задания в файле конфигурации Prometheus.

Примечание: Как уже упоминалось, использование непустого namespace или subsystem приведет к префиксу имени метрики. Вам может потребоваться внести соответствующие изменения, например, в определении панели на панели инструментов Grafana.

Настройка

Если вы считаете, что определенная метрика была упущена, вы можете легко расширить эту базовую реализацию. Лучший способ - использовать реестр Prometheus, используемый с методом ServeHTTP, и зарегистрировать метрики в соответствии с документацией клиента Prometheus здесь.

Краткий способ обновления этих метрик заключается в использовании декораторов. Полный исходный код можно найти на github.com/ThreeDotsLabs/watermill/message/decorator.go.

// ...
// MessageTransformSubscriberDecorator создает декоратор подписчика, который вызывает функцию преобразования для каждого сообщения, проходящего через подписчика.
func MessageTransformSubscriberDecorator(transform func(*Message)) SubscriberDecorator {
	if transform == nil {
		panic("функция преобразования пуста")
	}
	return func(sub Subscriber) (Subscriber, error) {
		return &messageTransformSubscriberDecorator{
			sub:       sub,
			transform: transform,
		}, nil
	}
}

// MessageTransformPublisherDecorator создает декоратор издателя, который вызывает функцию преобразования для каждого сообщения, проходящего через издателя.
func MessageTransformPublisherDecorator(transform func(*Message)) PublisherDecorator {
	if transform == nil {
		panic("функция преобразования пуста")
	}
	return func(pub Publisher) (Publisher, error) {
		return &messageTransformPublisherDecorator{
			Publisher: pub,
			transform: transform,
		}, nil
	}
}

type messageTransformSubscriberDecorator struct {
// ...

И/или использовать middleware в маршрутизаторах.

Более простой способ - обновлять необходимые метрики только в функциях обработчика.