tutorial

How to Monitor Apache Griffin Data Quality Health with Vigilmon

Apache Griffin Spark measure job failures, metric collection stalls, and scheduling gaps are invisible until downstream consumers discover that data quality SLAs were silently breached. Learn how to monitor Griffin data quality service availability, measure job liveness, and metric pipeline health with Vigilmon HTTP probes and heartbeat monitors.

Apache Griffin is an open-source data quality service platform for big data that runs accuracy, completeness, timeliness, validity, and uniqueness checks as Spark batch or streaming jobs. When a Griffin measure Spark job fails due to an executor OOM, the job silently exits with a non-zero status and Griffin marks the measure run as failed in its internal database — but no alert is sent to the data engineering team, so the next dashboard refresh shows a stale data quality score while analysts assume the pipeline is healthy; when Griffin's Livy job submission endpoint is unreachable due to a Yarn cluster resource shortage, new measure runs queue indefinitely in Griffin's scheduler without surfacing an actionable error; when Griffin's metric storage (Elasticsearch or InfluxDB) becomes unavailable, quality measurements complete but results are silently discarded — downstream consumers see a flatlined dashboard with no indication that ingestion failed. These are data quality SLA breaches disguised as normal system behavior.

Vigilmon gives you external visibility into Griffin's data quality service health through HTTP probe monitoring and heartbeat monitors. This tutorial covers both.

Why Griffin Needs External Monitoring

Griffin's failure modes span the Spark execution, metric storage, and API scheduling layers:

  • Spark measure job OOM failure: Griffin submits measure jobs to Spark via Livy; when an executor runs out of memory mid-execution, the job fails and Griffin logs the failure but does not alert consumers — data quality for that dimension is simply not measured for that window
  • Livy session pool exhaustion: Griffin uses Livy to manage Spark sessions; when all Livy sessions are occupied by long-running streaming measure jobs, new batch measure jobs queue and never execute — Griffin's scheduler shows them as pending indefinitely
  • Metric sink unavailability: Griffin writes measurement results to Elasticsearch, InfluxDB, or a JDBC sink; when the sink is unavailable during a measure run, Griffin completes the Spark job but discards the results silently — the measure is marked succeeded in Griffin's scheduler but no metric is ever written to the dashboard
  • Hive metastore connectivity failure: Griffin's accuracy measures compare a source dataset against a target dataset via Hive tables; when the Hive metastore is unreachable, the measure job throws MetaException and fails — but Griffin's API reports the measure as PENDING rather than FAILED, causing indefinite queue growth
  • Checkpoint corruption in streaming measures: Griffin streaming measures use Spark checkpoints for fault tolerance; a corrupt checkpoint directory after a cluster restart causes the streaming job to fail at startup and never recover — the streaming data quality dashboard shows stale metrics that are weeks old
  • Schedule drift in cron-based measures: Griffin uses a Quartz-based scheduler; after a Griffin service restart, Quartz can lose in-memory state for misfire-policy jobs and skip measure runs rather than catching up — data quality gaps accumulate across multiple windows before anyone notices

External monitoring with Vigilmon adds:

  • Proactive alerting when Griffin's service endpoint or metric APIs stop responding
  • Measure job liveness monitoring independent of Griffin's internal scheduler state
  • Heartbeat monitoring so you know when batch quality scans stop completing within their expected windows
  • Multi-region probe consensus that filters transient Livy session startup delays from genuine measure pipeline failures

Step 1: Build a Griffin Health Endpoint

Griffin 0.6+ ships a Spring Boot REST API. Use its built-in endpoints plus a custom health check that probes the Livy connection and metric sink.

Spring Boot Actuator (Griffin 0.6+)

Griffin's service module exposes /actuator/health by default. Enable component details in service/src/main/resources/application.properties:

management.endpoint.health.show-details=always
management.endpoints.web.exposure.include=health,info,metrics

The health endpoint includes Spring's default DB check (Griffin uses MySQL/PostgreSQL for measure metadata). Add a custom Livy connectivity check:

// health/LivyHealthIndicator.java
package com.yourorg.griffin.health;

import org.springframework.boot.actuate.health.Health;
import org.springframework.boot.actuate.health.HealthIndicator;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;

@Component
public class LivyHealthIndicator implements HealthIndicator {

    @Value("${livy.uri:http://localhost:8998}")
    private String livyUri;

    private final RestTemplate restTemplate = new RestTemplate();

    @Override
    public Health health() {
        try {
            String url = livyUri + "/sessions";
            restTemplate.getForObject(url, String.class);
            return Health.up()
                .withDetail("livy", livyUri)
                .withDetail("sessions_endpoint", "reachable")
                .build();
        } catch (Exception e) {
            return Health.down(e)
                .withDetail("livy", livyUri)
                .withDetail("sessions_endpoint", "unreachable")
                .build();
        }
    }
}

Custom Griffin Measure Status Endpoint

Expose a dedicated endpoint that checks for stale or accumulating measure failures:

// api/GriffinHealthResource.java
package com.yourorg.griffin.api;

import org.apache.griffin.core.job.JobServiceImpl;
import org.apache.griffin.core.job.entity.GriffinJob;
import org.springframework.web.bind.annotation.*;
import org.springframework.http.*;
import java.util.*;

@RestController
@RequestMapping("/api/v1/health")
public class GriffinHealthResource {

    private final JobServiceImpl jobService;

    public GriffinHealthResource(JobServiceImpl jobService) {
        this.jobService = jobService;
    }

    @GetMapping("/measures")
    public ResponseEntity<Map<String, Object>> measureHealth() {
        Map<String, Object> result = new LinkedHashMap<>();
        boolean healthy = true;

        try {
            List<GriffinJob> jobs = jobService.getAliveJobs();
            long errorJobs = jobs.stream()
                .filter(j -> "ERROR".equals(j.getJobStatus()))
                .count();

            result.put("total_jobs", jobs.size());
            result.put("error_jobs", errorJobs);

            if (errorJobs > 0) {
                result.put("status", "degraded");
                result.put("detail", errorJobs + " measure jobs in ERROR state");
                healthy = false;
            } else {
                result.put("status", "ok");
            }
        } catch (Exception e) {
            result.put("status", "down");
            result.put("error", e.getMessage());
            healthy = false;
        }

        return ResponseEntity.status(healthy ? 200 : 503).body(result);
    }
}

Node.js Sidecar

// health/griffin.js
const express = require('express');
const http = require('http');

const app = express();
const GRIFFIN_HOST = process.env.GRIFFIN_HOST || 'localhost';
const GRIFFIN_PORT = parseInt(process.env.GRIFFIN_PORT || '9090');
const LIVY_HOST = process.env.LIVY_HOST || 'localhost';
const LIVY_PORT = parseInt(process.env.LIVY_PORT || '8998');

function probe(host, port, path) {
  return new Promise((resolve, reject) => {
    const req = http.get({ host, port, path, timeout: 5000 }, res => {
      let body = '';
      res.on('data', c => { body += c; });
      res.on('end', () => resolve({ status: res.statusCode, body }));
    });
    req.on('error', reject);
    req.on('timeout', () => { req.destroy(); reject(new Error('timeout')); });
  });
}

app.get('/health/griffin', async (req, res) => {
  const checks = {};
  let healthy = true;

  try {
    const griffin = await probe(GRIFFIN_HOST, GRIFFIN_PORT, '/actuator/health');
    checks.griffin_api = griffin.status === 200 ? 'ok' : 'down';
    if (griffin.status !== 200) healthy = false;
  } catch (e) {
    checks.griffin_api = 'down';
    healthy = false;
  }

  try {
    const livy = await probe(LIVY_HOST, LIVY_PORT, '/sessions');
    checks.livy = livy.status === 200 ? 'ok' : 'down';
    if (livy.status !== 200) healthy = false;
  } catch (e) {
    checks.livy = 'down';
    healthy = false;
  }

  return res.status(healthy ? 200 : 503).json({ status: healthy ? 'ok' : 'down', checks });
});

app.listen(3020);

Python Sidecar

# health_griffin.py
import os, requests
from flask import Flask, jsonify

app = Flask(__name__)

GRIFFIN_BASE = f"http://{os.environ.get('GRIFFIN_HOST','localhost')}:{os.environ.get('GRIFFIN_PORT','9090')}"
LIVY_BASE = f"http://{os.environ.get('LIVY_HOST','localhost')}:{os.environ.get('LIVY_PORT','8998')}"

@app.route('/health/griffin')
def health():
    checks = {}
    healthy = True

    try:
        r = requests.get(f'{GRIFFIN_BASE}/actuator/health', timeout=5)
        checks['griffin_api'] = 'ok' if r.status_code == 200 else 'down'
        if r.status_code != 200:
            healthy = False
    except Exception as e:
        checks['griffin_api'] = f'down: {e}'
        healthy = False

    try:
        r = requests.get(f'{LIVY_BASE}/sessions', timeout=5)
        checks['livy'] = 'ok' if r.status_code == 200 else 'down'
        if r.status_code != 200:
            healthy = False
    except Exception as e:
        checks['livy'] = f'down: {e}'
        healthy = False

    return jsonify({'status': 'ok' if healthy else 'down', 'checks': checks}), 200 if healthy else 503

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=3020)

Step 2: Configure Vigilmon HTTP Monitor for Griffin

  1. Log in to vigilmon.online and go to Monitors → New Monitor
  2. Choose HTTP / HTTPS
  3. Set the URL to Griffin's health endpoint: https://griffin.your-org.example.com/actuator/health
  4. Set the check interval to 1 minute
  5. Under Expected response, configure:
    • Status code: 200
    • Response body contains: "status":"UP"
    • Response time threshold: 5000ms
  6. Under Alert channels, assign your Slack or PagerDuty channel
  7. Save the monitor

Add monitors for Griffin's API and Livy:

| Endpoint | What it catches | |---|---| | /actuator/health | Griffin service, DB metadata store, disk space | | /api/v1/health/measures | Measure job accumulation, ERROR state jobs | | http://livy:8998/sessions | Livy session pool availability | | /api/v1/griffin/metrics/values?metricName=accuracy | Metric sink write availability |

For the Livy monitor, point Vigilmon at an internal load-balanced address or use the sidecar health proxy if Livy is not publicly accessible.

Vigilmon's multi-region probing filters transient Livy session warmup delays from genuine Spark cluster resource exhaustion.

Step 3: Heartbeat Monitoring for Batch Measure Jobs

HTTP health checks verify Griffin service and Livy availability but not whether individual data quality measure runs are completing. Griffin can report healthy via Actuator while Quartz has drifted and skipped the last six measure windows due to a misfire policy misconfiguration.

Vigilmon heartbeat monitors catch measure pipeline gaps.

Set Up Heartbeat Monitors

For hourly accuracy measure liveness:

  1. In Vigilmon, go to Monitors → New Monitor → Heartbeat
  2. Set the name: griffin-accuracy-measure
  3. Set the expected interval: 70 minutes
  4. Set the grace period: 90 minutes
  5. Save — copy the heartbeat URL

For daily completeness measure liveness:

  1. Create another Heartbeat monitor named griffin-completeness-daily
  2. Set the expected interval: 25 hours
  3. Set the grace period: 27 hours
  4. Save — copy the heartbeat URL

Wire Heartbeats Into Griffin Measure Jobs

Add a post-execution heartbeat ping to Griffin's job completion hooks via a custom Quartz job listener:

// quartz/GriffinHeartbeatJobListener.java
package com.yourorg.griffin.quartz;

import org.quartz.*;
import java.net.URI;
import java.net.http.*;
import java.util.Map;

public class GriffinHeartbeatJobListener implements JobListener {

    private final Map<String, String> jobHeartbeatUrls;
    private final HttpClient http = HttpClient.newHttpClient();

    public GriffinHeartbeatJobListener(Map<String, String> jobHeartbeatUrls) {
        this.jobHeartbeatUrls = jobHeartbeatUrls;
    }

    @Override
    public String getName() { return "GriffinHeartbeatListener"; }

    @Override
    public void jobWasExecuted(JobExecutionContext context, JobExecutionException ex) {
        if (ex != null) return; // do not ping on failure
        String jobName = context.getJobDetail().getKey().getName();
        String url = jobHeartbeatUrls.get(jobName);
        if (url == null) return;
        try {
            http.sendAsync(
                HttpRequest.newBuilder(URI.create(url)).GET().build(),
                HttpResponse.BodyHandlers.discarding()
            );
        } catch (Exception ignored) {}
    }

    @Override
    public void jobToBeExecuted(JobExecutionContext context) {}

    @Override
    public void jobExecutionVetoed(JobExecutionContext context) {}
}

For Griffin measure jobs submitted externally via the REST API, wrap the submission with a post-completion heartbeat in your pipeline orchestrator:

# pipeline/griffin_measure_runner.py
import os, requests, time

GRIFFIN_BASE = os.environ['GRIFFIN_BASE']
HEARTBEAT_URL = os.environ.get('VIGILMON_GRIFFIN_MEASURE_HEARTBEAT_URL')

def run_measure(job_name, cron=None):
    # Trigger Griffin measure run via REST API
    resp = requests.post(
        f'{GRIFFIN_BASE}/api/v1/jobs/triggerJob',
        json={'jobName': job_name},
        timeout=30,
    )
    resp.raise_for_status()

    # Poll until complete
    job_id = resp.json()['jobId']
    for _ in range(120):
        status_resp = requests.get(
            f'{GRIFFIN_BASE}/api/v1/jobs/{job_id}', timeout=10
        )
        status = status_resp.json().get('jobStatus')
        if status == 'COMPLETE':
            if HEARTBEAT_URL:
                requests.get(HEARTBEAT_URL, timeout=5)
            return
        elif status in ('ERROR', 'FAILED'):
            raise RuntimeError(f'Griffin measure {job_name} failed with status {status}')
        time.sleep(30)

    raise TimeoutError(f'Griffin measure {job_name} did not complete within 60 minutes')

if __name__ == '__main__':
    run_measure('accuracy_measure_hourly')

Step 4: Alert Routing for Griffin Failures

Griffin failures range from a single measure job failing due to a Spark executor crash to the entire Livy session pool being exhausted. Route alerts by data quality SLA impact:

| Monitor | Alert Channel | Priority | |---|---|---| | HTTP: /actuator/health Griffin service | Slack + PagerDuty | P1 | | HTTP: Livy sessions endpoint | Slack + PagerDuty | P1 | | HTTP: measure error accumulation endpoint | Slack + PagerDuty | P1 | | HTTP: metric sink write check | Slack | P2 | | Heartbeat: hourly accuracy measure | Slack + PagerDuty | P1 | | Heartbeat: daily completeness measure | Slack + email | P2 | | Heartbeat: streaming data quality job | Slack + PagerDuty | P1 |

Set response time thresholds as early warning signals:

  • Alert at 3000ms for the Griffin API health check (slow responses indicate metadata DB pressure)
  • Alert at 5000ms for the Livy sessions endpoint (slow responses indicate session pool pressure)
  • Alert at 90 minutes gap in hourly accuracy measure heartbeat (one full window missed)
  • Alert at 27 hours gap in daily completeness heartbeat

For production Griffin deployments serving downstream data pipelines, set up a status page in Vigilmon showing data quality service availability and measure job liveness — data consumers and data stewards can check this independently rather than pinging the data engineering team every time a dashboard shows stale quality scores.

Summary

Griffin failures typically appear as stale data quality scores on dashboards rather than hard service outages — the API layer can be healthy while Spark measure jobs are silently failing or Quartz has drifted past the last scheduled window. External monitoring catches these before downstream teams discover a data quality SLA has been silently breached:

| Monitor Type | What It Covers | |---|---| | HTTP monitor on /actuator/health | Griffin service, DB metadata, Livy reachability | | HTTP monitor on measure error endpoint | Accumulating ERROR-state measure jobs | | HTTP monitor on Livy sessions | Spark session pool availability | | Heartbeat monitor (batch measures) | End-to-end measure job completion within SLA window | | Heartbeat monitor (streaming measures) | Continuous streaming data quality pipeline liveness |

Get started free at vigilmon.online — your first Griffin data quality monitor is running in under two minutes.

Monitor your app with Vigilmon

Free plan — 5 monitors, no credit card required. Up and running in 60 seconds.

Start free →
← Back to all articles