Add Elasticsearch index-freshness check type

Author: brvier

.gitignore

@@ -3,3 +3,4 @@
 *.db-wal
 *.db-shm
 gjallar.yaml
+create-gjallar-user-mediainsights.sql

CHANGELOG.md

@@ -3,6 +3,14 @@
 All notable changes to Gjallar are documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
+## [0.7.0] - 2026-07-06
+
+### Added
+
+- Elasticsearch check type (`type: elasticsearch`): index freshness measured
+  as hours since `max(timestamp_field)`, evaluated against a rule (e.g. "< 3").
+  Fields: `url`, `index`, `timestamp_field`, `rule`.
+
 ## [0.6.0] - 2026-07-06
 
 ### Added

README.md

@@ -16,6 +16,7 @@ HTMX-refreshed), and alerts you when something goes down — and when it recover
 | `oracle` | SQL query result against a rule (pure-Go go-ora driver, no Oracle client needed) |
 | `ping` | ICMP echo (privileged or unprivileged) |
 | `redis` | TCP connect + optional `AUTH` + `PING`/`+PONG` |
+| `elasticsearch` | freshness: hours since `max(timestamp_field)` in an index, checked against a rule |
 | `prometheus` | Fetches a `/metrics` route and evaluates a rule against metric values, with optional label selectors |
 
 Rules: `> N`, `>= N`, `< N`, `<= N`, `== x`, `!= x`, `~ regex`, `rows > 0` (row count).

gjallar.example.yaml

@@ -96,6 +96,14 @@ monitors:
     port: 6379            # default 6379
     password: "${REDIS_PASSWORD}"   # omit if no AUTH
 
+  # --- Elasticsearch index freshness: hours since max(timestamp_field) ---
+  - name: es-broadcasts
+    type: elasticsearch
+    url: "http://es-host:9200"
+    index: "broadcasts"
+    timestamp_field: "start_date"
+    rule: "< 3"           # alert if the newest document is more than 3h old
+
   # --- Prometheus metrics: rule must hold for every matching series ---
   - name: node1-disk
     type: prometheus

internal/check/check.go

@@ -41,6 +41,8 @@ func New(m config.Monitor) (Checker, error) {
 		c, err = newPingCheck(m)
 	case "redis":
 		c, err = newRedisCheck(m)
+	case "elasticsearch":
+		c, err = newESCheck(m)
 	case "prometheus":
 		c, err = newPromCheck(m)
 	default:

internal/check/elasticsearch.go

@@ -0,0 +1,84 @@
+package check
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"strconv"
+	"strings"
+	"time"
+
+	"gjallar/internal/config"
+)
+
+// esCheck measures the freshness of an Elasticsearch index: the number of hours
+// between now and max(timestamp_field), evaluated against the rule (e.g. "< 3").
+// This is the end-of-chain freshness signal — when an indexing worker stalls,
+// max(timestamp_field) stops advancing and the lag grows without bound.
+type esCheck struct {
+	baseURL string
+	index   string
+	tsField string
+	rule    *Rule
+	client  *http.Client
+}
+
+func newESCheck(m config.Monitor) (*esCheck, error) {
+	rule, err := ParseRule(m.Rule)
+	if err != nil {
+		return nil, err
+	}
+	if rule.TargetsRows() {
+		return nil, fmt.Errorf("rule %q: row count rules do not apply to elasticsearch", m.Rule)
+	}
+	return &esCheck{
+		baseURL: strings.TrimRight(m.URL, "/"),
+		index:   m.Index,
+		tsField: m.TimestampField,
+		rule:    rule,
+		client:  &http.Client{},
+	}, nil
+}
+
+func (c *esCheck) Check(ctx context.Context) (bool, string) {
+	body := fmt.Sprintf(`{"size":0,"aggs":{"max_ts":{"max":{"field":%q}}}}`, c.tsField)
+	url := c.baseURL + "/" + c.index + "/_search"
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, strings.NewReader(body))
+	if err != nil {
+		return false, err.Error()
+	}
+	req.Header.Set("Content-Type", "application/json")
+
+	resp, err := c.client.Do(req)
+	if err != nil {
+		return false, err.Error()
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode < 200 || resp.StatusCode > 299 {
+		return false, fmt.Sprintf("status %d querying %s", resp.StatusCode, c.index)
+	}
+
+	var parsed struct {
+		Aggregations struct {
+			MaxTS struct {
+				Value float64 `json:"value"` // epoch millis
+			} `json:"max_ts"`
+		} `json:"aggregations"`
+	}
+	if err := json.NewDecoder(resp.Body).Decode(&parsed); err != nil {
+		return false, fmt.Sprintf("parsing response: %v", err)
+	}
+	ms := parsed.Aggregations.MaxTS.Value
+	if ms <= 0 {
+		return false, fmt.Sprintf("no %q value (empty index?)", c.tsField)
+	}
+
+	last := time.UnixMilli(int64(ms))
+	lagHours := time.Since(last).Hours()
+	if err := c.rule.EvalValue(strconv.FormatFloat(lagHours, 'f', 2, 64)); err != nil {
+		return false, fmt.Sprintf("freshness lag %.1fh (last %s): %v",
+			lagHours, last.UTC().Format("2006-01-02 15:04 MST"), err)
+	}
+	return true, ""
+}

internal/check/elasticsearch_test.go

@@ -0,0 +1,71 @@
+package check
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"gjallar/internal/config"
+)
+
+func esMonitor(url, rule string) config.Monitor {
+	return config.Monitor{
+		Name: "t", Type: "elasticsearch", URL: url, Index: "broadcasts",
+		TimestampField: "start_date", Rule: rule,
+		Timeout: config.Duration(5 * time.Second),
+	}
+}
+
+// esServer returns a fake ES that reports max(start_date) = now - lag.
+func esServer(t *testing.T, lag time.Duration) *httptest.Server {
+	t.Helper()
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if !strings.HasSuffix(r.URL.Path, "/broadcasts/_search") {
+			w.WriteHeader(http.StatusNotFound)
+			return
+		}
+		ms := time.Now().Add(-lag).UnixMilli()
+		fmt.Fprintf(w, `{"aggregations":{"max_ts":{"value":%d.0}}}`, ms)
+	}))
+	t.Cleanup(srv.Close)
+	return srv
+}
+
+func TestESCheckFresh(t *testing.T) {
+	srv := esServer(t, 30*time.Minute) // 0.5h lag
+	c, _ := newESCheck(esMonitor(srv.URL, "< 3"))
+	if ok, msg := c.Check(context.Background()); !ok {
+		t.Errorf("expected ok, got %q", msg)
+	}
+}
+
+func TestESCheckStale(t *testing.T) {
+	srv := esServer(t, 13*time.Hour) // the incident: 13h lag
+	c, _ := newESCheck(esMonitor(srv.URL, "< 3"))
+	ok, msg := c.Check(context.Background())
+	if ok || !strings.Contains(msg, "freshness lag") {
+		t.Errorf("got ok=%v msg=%q", ok, msg)
+	}
+}
+
+func TestESCheckEmpty(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		fmt.Fprint(w, `{"aggregations":{"max_ts":{"value":null}}}`)
+	}))
+	defer srv.Close()
+	c, _ := newESCheck(esMonitor(srv.URL, "< 3"))
+	if ok, msg := c.Check(context.Background()); ok || !strings.Contains(msg, "empty index") {
+		t.Errorf("got ok=%v msg=%q", ok, msg)
+	}
+}
+
+func TestESCheckDown(t *testing.T) {
+	c, _ := newESCheck(esMonitor("http://127.0.0.1:1", "< 3"))
+	if ok, _ := c.Check(context.Background()); ok {
+		t.Error("expected failure")
+	}
+}

internal/config/config.go

@@ -91,10 +91,15 @@ type Monitor struct {
 	// prometheus
 	Metric string            `yaml:"metric"`
 	Labels map[string]string `yaml:"labels"`
+
+	// elasticsearch
+	Index          string `yaml:"index"`
+	TimestampField string `yaml:"timestamp_field"` // freshness = hours since max(this field)
 }
 
 var monitorTypes = map[string]bool{
 	"http": true, "postgres": true, "oracle": true, "ping": true, "prometheus": true, "redis": true,
+	"elasticsearch": true,
 }
 
 // envRe matches ${VAR} only; a bare $ is left alone so regex rules like
@@ -238,7 +243,7 @@ func (c *Config) Validate() error {
 
 func (m *Monitor) validate() error {
 	if !monitorTypes[m.Type] {
-		return fmt.Errorf("unknown type %q (supported: http, postgres, oracle, ping, prometheus, redis)", m.Type)
+		return fmt.Errorf("unknown type %q (supported: http, postgres, oracle, ping, prometheus, redis, elasticsearch)", m.Type)
 	}
 	switch m.Type {
 	case "http":
@@ -274,6 +279,19 @@ func (m *Monitor) validate() error {
 		if m.Rule == "" {
 			return fmt.Errorf("rule is required")
 		}
+	case "elasticsearch":
+		if m.URL == "" {
+			return fmt.Errorf("url is required")
+		}
+		if m.Index == "" {
+			return fmt.Errorf("index is required")
+		}
+		if m.TimestampField == "" {
+			return fmt.Errorf("timestamp_field is required")
+		}
+		if m.Rule == "" {
+			return fmt.Errorf("rule is required")
+		}
 	}
 	return nil
 }

test.sh

@@ -0,0 +1 @@
+for h in 10.2.70.100 10.2.70.101 10.2.70.103; do ssh -t yacuser@$h "sudo pg_enc -m -k /etc/pgpool2/.pgpoolkey -u gjallar '$(grep -m1 PASSWORD create-gjallar-user-mediainsights.sql | sed "s/.*PASSWORD '//;s/'.*//")' && sudo systemctl reload pgpool2"; done