ConnectMon

A tool for monitoring tasks and connectors in Kafka Connect.

Installation

With pip

pip install connectmon

With poetry

poetry add connectmon

Usage

from connectmon import env, API
from connectmon.messaging import TeamsService


## Setup Kafka Connect Rest API client and check if cluster is reachable
connect = API(env.CONNECT_URL)

## Get all connectors and check if any are in a failed state
connectors = connect.get_all_connectors()

## Loop through all channels and send messages
for channel in env.CHANNELS.channels:
    if channel.type == "teams":
        service = TeamsService(connect, channel)
        service.process_channel_connectors(connectors)
        service.send_message()

Configuration

Base configuration for ConnectMon is handled through Environment Variables.

Environment Variables

Name	Type	Description	Default
CONFIG_PATH	Optional string	The path to the channel configuration file	""
CONNECT_URL	string	The URL of the Connect cluster	"http://localhost:8083"
SKIP_TLS_VERIFY	boolean	Whether to skip TLS verification	False
ENVIRONMENT	string	The environment the application is running in	"dev"
LOG_LEVEL	string	The log level for the application	"INFO"
LOG_FORMAT	string	The log format for the application	"(asctime)s - ..."

Name in bold are required!

If CONFIG_PATH is provided, the application will attempt to load configuration from the file, which will set settings.CHANNELS with the channels configured in the supplied configuration file.

Channel Configuration

You can configure specific channels to receive notifications when connectors or tasks are paused or failed.

A config file could look like this

channels:
  - name: my-teams-team-name
    type: teams
    url: https://my-org.webhook.office.com/webhookb2/...
    actions:
      - RESTART_FAILED
    include:
      - i-want-to-monitor-this-connector
      - this-too
    exclude:
      - who-cares-about-this-connector
      - this-is-someone-elses-problem

Supported fields for channels are:

Name	Type	Description	Default
name	string	Name of the channel	""
type	string	Type of channel	""
url	string	Url to send payload to	""
actions	list of strings	Can be any of RESTART_FAILED, RESTART_FAILED_CONNECTORS, RESTART_FAILED_TASKS, RESUME_PAUSED_CONNECTORS	RESTART_FAILED
include	list of strings	Names of connectors to include for this channel	["*"]
exclude	list of strings	Names of connectors to exclude for this channel	[]

Current only Microsoft Teams or type: "teams" is supported.