Lenses.io is an enterprise Kafka DataOps and observability platform that connects to your existing Apache Kafka clusters — self-hosted, Confluent Cloud, Amazon MSK, or Azure Event Hubs — and provides data exploration, consumer group lag monitoring, schema management, connector management, data masking policies, and a visual topology view. Lenses doesn't replace Kafka; it's the control plane that makes Kafka observable and governable for your entire team. But when Lenses itself goes down, or loses connectivity to Kafka, your team loses visibility into your entire streaming infrastructure. Vigilmon lets you monitor the monitor — ensuring Lenses stays healthy, connected, and able to serve your DataOps team.
What You'll Set Up
- Lenses application process health monitoring
- Lenses-to-Kafka broker connectivity checks
- Schema Registry connectivity monitoring
- Kafka Connect REST API connectivity monitoring
- Consumer group lag data freshness alerting
- LSQL query execution health monitoring
- Data policy enforcement health checks
- Lenses audit log write health monitoring
- Authentication success rate monitoring
- Lenses web UI availability checks
Prerequisites
- Lenses deployed as a Docker container (Lenses Box) or on-premise JVM process
- Lenses configured and connected to at least one Kafka cluster
- Lenses API access (Lenses exposes a REST API on the same port as the UI)
- A free Vigilmon account
Why Monitor Lenses?
Lenses is itself a monitoring and observability tool — but it's not self-monitoring. When Lenses crashes, your team loses the only dashboard showing consumer group lag across your Kafka cluster. When Lenses loses Kafka connectivity, consumer group lag data goes stale without any visible error. When Schema Registry connectivity drops, schema browsing and data masking policy enforcement may silently fail. And when LSQL queries start failing, the SQL Studio your team uses for Kafka data exploration becomes unavailable. Vigilmon's independent health checks catch all of these failure modes from outside Lenses, giving you an alert path that doesn't depend on Lenses being up.
Step 1: Monitor Lenses Application Health
Lenses exposes a health endpoint at the root of its API:
GET http://<lenses-host>:9991/api/health
(The default Lenses port is 9991, but this may be configured differently in your deployment. Check your lenses.conf or Docker environment for LENSES_PORT.)
The health endpoint returns 200 OK with a JSON body when Lenses is running. Add a Vigilmon monitor:
- Click Add Monitor → HTTP / HTTPS.
- Enter
http://<lenses-host>:9991/api/health. - Set Expected HTTP status to
200. - Set Check interval to
1 minute. - Click Save.
This is your primary Lenses liveness check. If Lenses crashes or the container exits, this check fails within 1 minute and your team is alerted.
For a richer health check, authenticate with the Lenses API and call a lightweight authenticated endpoint:
# Get a Lenses API token
curl -X POST http://<lenses-host>:9991/api/login \
-H "Content-Type: application/json" \
-d '{"user": "<username>", "password": "<password>"}'
Store the returned token and use it for Step 2 onwards.
Step 2: Monitor Lenses-to-Kafka Broker Connectivity
Lenses connects to Kafka brokers to read topic metadata, consumer group offsets, and messages. Loss of Kafka connectivity means topic data, consumer group lag, and topology views all stop updating.
Lenses exposes connection status via its API:
GET http://<lenses-host>:9991/api/kafka/brokers
Authorization: Bearer <token>
Write a probe that checks Lenses returns a non-empty broker list and all brokers are reachable:
import os
import json
import requests
from http.server import BaseHTTPRequestHandler, HTTPServer
LENSES_HOST = os.environ.get("LENSES_HOST", "localhost")
LENSES_PORT = int(os.environ.get("LENSES_PORT", "9991"))
LENSES_TOKEN = os.environ["LENSES_TOKEN"]
BASE_URL = f"http://{LENSES_HOST}:{LENSES_PORT}/api"
HEADERS = {"Authorization": f"Bearer {LENSES_TOKEN}"}
def check_kafka_connectivity():
resp = requests.get(f"{BASE_URL}/kafka/brokers", headers=HEADERS, timeout=10)
resp.raise_for_status()
brokers = resp.json()
online = [b for b in brokers if b.get("online", False)]
return len(online) > 0, {"online_brokers": len(online), "total_brokers": len(brokers)}
class KafkaConnHandler(BaseHTTPRequestHandler):
def do_GET(self):
if self.path == "/kafka-connectivity":
try:
healthy, info = check_kafka_connectivity()
status = 200 if healthy else 503
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps(info).encode())
except Exception as e:
self.send_response(503)
self.end_headers()
self.wfile.write(json.dumps({"error": str(e)}).encode())
else:
self.send_response(404)
self.end_headers()
def log_message(self, format, *args):
pass
HTTPServer(("0.0.0.0", 8080), KafkaConnHandler).serve_forever()
Deploy this probe and add a Vigilmon monitor at http://<probe-host>:8080/kafka-connectivity with a 1-minute interval. If Lenses loses Kafka connectivity, all consumer group lag data goes stale and topology views stop updating.
Step 3: Monitor Schema Registry Connectivity
Lenses connects to Schema Registry for schema browsing and data masking policy enforcement. Loss of Schema Registry connectivity breaks schema exploration in the Lenses UI.
def check_schema_registry():
resp = requests.get(f"{BASE_URL}/registry/schemas", headers=HEADERS, timeout=10)
resp.raise_for_status()
schemas = resp.json()
# A successful response (even an empty list) confirms SR connectivity
return True, {"schema_count": len(schemas) if isinstance(schemas, list) else 0}
Expose /schema-registry-health on port 8081. Return 503 on any request error (Lenses itself reports Schema Registry connectivity failures as API errors). Monitor with Vigilmon at 2-minute intervals.
Step 4: Monitor Kafka Connect Connectivity
Lenses manages Kafka Connect clusters via their REST API. If Lenses loses connectivity to the Kafka Connect REST API, connector status in the Lenses UI becomes stale and connector management operations fail.
def check_connect_health():
resp = requests.get(f"{BASE_URL}/connectors", headers=HEADERS, timeout=10)
resp.raise_for_status()
connectors = resp.json()
# If Lenses can't reach Connect, it returns empty or errors
# Check for a successful non-error response
failed = [c for c in (connectors if isinstance(connectors, list) else [])
if c.get("status", {}).get("connector", {}).get("state") == "FAILED"]
return len(failed) == 0, {
"total_connectors": len(connectors) if isinstance(connectors, list) else 0,
"failed_connectors": len(failed)
}
Expose /connect-health on port 8082. Return 503 when any connector is in FAILED state (indicating Lenses can reach Connect, but a connector is broken) or when the API call itself fails (Lenses can't reach Connect at all). Monitor with Vigilmon.
Step 5: Monitor Consumer Group Lag Data Freshness
One of Lenses' primary features is consumer group lag monitoring. If Lenses' lag data becomes stale — because Lenses has lost Kafka connectivity or the lag refresh job has failed — teams are making decisions on outdated data without knowing it.
import datetime
LAG_FRESHNESS_SLA_SECONDS = int(os.environ.get("LAG_FRESHNESS_SLA_SECONDS", "300"))
def check_lag_freshness():
resp = requests.get(f"{BASE_URL}/groups", headers=HEADERS, timeout=10)
resp.raise_for_status()
groups = resp.json()
stale = []
now = datetime.datetime.now(datetime.timezone.utc)
for group in groups if isinstance(groups, list) else []:
last_updated = group.get("lastUpdated")
if last_updated:
last_dt = datetime.datetime.fromisoformat(last_updated.replace("Z", "+00:00"))
age_seconds = (now - last_dt).total_seconds()
if age_seconds > LAG_FRESHNESS_SLA_SECONDS:
stale.append({
"group": group.get("id"),
"age_seconds": round(age_seconds)
})
return len(stale) == 0, {"stale_groups": stale, "total_groups": len(groups) if isinstance(groups, list) else 0}
Expose /lag-freshness on port 8083. Return 503 when any consumer group's lag data is older than 5 minutes. Monitor with Vigilmon at 2-minute intervals.
Step 6: Monitor LSQL Query Health
Lenses SQL (LSQL) lets teams explore Kafka topics with SQL queries in real time. LSQL query execution failures block data exploration and may indicate Lenses-Kafka connectivity degradation.
Probe LSQL by running a trivial query against a known topic:
PROBE_TOPIC = os.environ.get("LSQL_PROBE_TOPIC", "")
def check_lsql_health():
if not PROBE_TOPIC:
return True, {"skipped": "no probe topic configured"}
payload = {
"sql": f"SELECT * FROM `{PROBE_TOPIC}` LIMIT 1",
"stats": 1
}
resp = requests.post(
f"{BASE_URL}/sql/execute",
headers={**HEADERS, "Content-Type": "application/json"},
json=payload,
timeout=15
)
success = resp.status_code in (200, 201)
return success, {"status_code": resp.status_code}
Expose /lsql-health on port 8084. Return 503 on LSQL execution failure. Use a low-volume topic for the probe query to avoid noise. Monitor with Vigilmon at 5-minute intervals.
Step 7: Monitor Data Policy Enforcement Health
Lenses data masking policies protect sensitive Kafka data from unauthorized exposure. Policy enforcement failures mean sensitive fields are being returned unmasked to unauthorized consumers.
def check_policy_health():
resp = requests.get(f"{BASE_URL}/policies", headers=HEADERS, timeout=10)
resp.raise_for_status()
policies = resp.json()
# Check that policies exist and none are in an error state
if not isinstance(policies, list):
return False, {"error": "unexpected response format"}
errored = [p for p in policies if p.get("status") == "ERROR"]
return len(errored) == 0, {
"total_policies": len(policies),
"errored_policies": [p.get("name") for p in errored]
}
Expose /policy-health on port 8085. Return 503 when any data masking policy is in an error state. Alert on 503 — failed data policies mean sensitive Kafka data may be exposed. Monitor at 5-minute intervals.
Step 8: Monitor the Lenses Audit Log
Lenses logs all user actions (topic browsing, SQL queries, connector changes) to an audit log for compliance. If the audit log write fails, you have a compliance gap that may be invisible to your team.
def check_audit_log():
# Query recent audit events; a successful response confirms write health
resp = requests.get(
f"{BASE_URL}/audit",
headers=HEADERS,
params={"limit": 1},
timeout=10
)
resp.raise_for_status()
events = resp.json()
has_events = len(events) > 0 if isinstance(events, list) else False
return True, {"recent_events": len(events) if isinstance(events, list) else 0}
Expose /audit-health on port 8086. Return 503 when the audit API returns an error (indicating write problems). Monitor with Vigilmon at 5-minute intervals.
Step 9: Monitor Lenses Authentication Health
Lenses supports LDAP, SSO (SAML, OIDC), and local authentication. Authentication failures block the entire team's access to the Kafka observability control plane.
Probe authentication health by attempting a login with a dedicated monitoring service account:
MONITOR_USER = os.environ["LENSES_MONITOR_USER"]
MONITOR_PASS = os.environ["LENSES_MONITOR_PASS"]
def check_auth_health():
resp = requests.post(
f"{BASE_URL}/login",
json={"user": MONITOR_USER, "password": MONITOR_PASS},
timeout=10
)
success = resp.status_code == 200 and "token" in resp.json()
return success, {"auth_status": resp.status_code}
Expose /auth-health on port 8087. Return 503 on authentication failure. Use a dedicated read-only monitoring service account (not a personal account) to avoid locking out real users. Monitor with Vigilmon at 5-minute intervals.
Step 10: Monitor the Lenses Web UI
The Lenses web UI is the primary interface for your DataOps team. Even if the Lenses API is healthy, a UI delivery failure (nginx misconfiguration, static asset missing) blocks team access.
Add a Vigilmon monitor directly on the Lenses web port:
- Click Add Monitor → HTTP / HTTPS.
- Enter
http://<lenses-host>:9991/(or your configured URL/domain for Lenses). - Set Expected HTTP status to
200. - Optionally set Expected body contains to
Lensesto verify the correct page is served. - Set Check interval to
1 minute. - Click Save.
This independently verifies that the Lenses UI is reachable and loading, separate from the API health check in Step 1.
Alerting Configuration
| Monitor | Condition | Action |
|---------|-----------|--------|
| Lenses process health | Any 503 | Page on-call immediately |
| Kafka broker connectivity | Any 503 | Page on-call immediately |
| Schema Registry connectivity | Any 503 | Notify data engineering team |
| Kafka Connect connectivity | Any 503 | Notify data engineering team |
| Consumer group lag freshness | 503 (lag data > 5 min old) | Notify data engineering team |
| LSQL query health | Any 503 | Notify data engineering team |
| Data policy enforcement | Any 503 | Page on-call + security team immediately |
| Audit log health | Any 503 | Notify compliance team |
| Authentication health | Any 503 | Page on-call immediately |
| Lenses web UI | Any 503 | Page on-call immediately |
Route alerts to Slack, PagerDuty, or email using Vigilmon's Notification Channels. The data policy enforcement and authentication checks warrant the highest alert priority — failures directly impact security and team access.
Conclusion
Lenses.io is your team's window into Kafka — which means monitoring Lenses itself is as important as monitoring Kafka. By adding a lightweight Python probe that polls the Lenses REST API and wiring it into Vigilmon, you get independent visibility into the ten critical health dimensions: Lenses process liveness, Kafka broker connectivity, Schema Registry availability, Kafka Connect management, consumer group lag data freshness, LSQL query health, data masking policy enforcement, audit log integrity, authentication health, and UI availability. When Lenses has a problem, you'll know in under a minute — before your team discovers it the hard way.
Get started free at vigilmon.online.