Elasticsearch health check

Taskfile.yaml

@@ -54,6 +54,8 @@ tasks:
         sh: docker-compose port cassandra 9042
       NATS_HOST:
         sh: docker-compose port nats 4222
+      ELASTICSEARCH_HOST:
+        sh: docker-compose port elasticsearch 9200
 
     env:
       HEALTH_GO_PG_PQ_DSN: 'postgres://test:test@{{.PG_PQ_HOST}}/test?sslmode=disable'
@@ -68,3 +70,4 @@ tasks:
       HEALTH_GO_INFLUXDB_URL: 'http://{{.INFLUX_HOST}}'
       HEALTH_GO_CASSANDRA_HOST: '{{.CASSANDRA_HOST}}'
       HEALTH_GO_NATS_DSN: 'nats://{{.NATS_HOST}}'
+      HEALTH_GO_ES_DSN: '{{.ELASTICSEARCH_HOST}}'

checks/elasticsearch/check.go

@@ -0,0 +1,117 @@
+package elasticsearch
+
+import (
+	"context"
+	"crypto/tls"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"time"
+)
+
+// Config is the Elasticsearch checker configuration settings container.
+type Config struct {
+	DSN         string // DSN is the Elasticsearch connection DSN. Required.
+	Password    string // Password is the Elasticsearch connection password. Required.
+	SSLCertPath string // SSLCertPath is the path to the SSL certificate to use for the connection. Optional.
+}
+
+// New creates a new Elasticsearch health check that verifies the status of the cluster.
+func New(config Config) func(ctx context.Context) error {
+	if config.DSN == "" || config.Password == "" {
+		return func(ctx context.Context) error {
+			return fmt.Errorf("elasticsearch DSN and password are required")
+		}
+	}
+
+	client, err := makeHTTPClient(config.SSLCertPath)
+	if err != nil {
+		return func(ctx context.Context) error {
+			return fmt.Errorf("failed to create Elasticsearch HTTP client: %w", err)
+		}
+	}
+
+	return func(ctx context.Context) error {
+		return checkHealth(ctx, client, config.DSN, config.Password)
+	}
+}
+
+func makeHTTPClient(sslCertPath string) (*http.Client, error) {
+	httpClient := http.Client{
+		Timeout: 5 * time.Second,
+	}
+
+	// If SSLCert is set, configure the client to use it.
+	// Otherwise, skip TLS verification.
+
+	if sslCertPath != "" {
+		cert, err := tls.LoadX509KeyPair(sslCertPath, sslCertPath)
+		if err != nil {
+			return nil, fmt.Errorf("failed to load Elasticsearch SSL certificate: %w", err)
+		}
+
+		// Configure the client to use the certificate.
+		httpTransport := &http.Transport{
+			TLSClientConfig: &tls.Config{
+				Certificates: []tls.Certificate{cert},
+			},
+		}
+
+		httpClient.Transport = httpTransport
+		return &httpClient, nil
+	}
+
+	// Configure the client to skip TLS verification.
+	httpTransport := &http.Transport{
+		TLSClientConfig: &tls.Config{
+			InsecureSkipVerify: true,
+		},
+	}
+
+	httpClient.Transport = httpTransport
+
+	return &httpClient, nil
+}
+
+func checkHealth(ctx context.Context, client *http.Client, dsn string, password string) error {
+	req, err := http.NewRequestWithContext(
+		ctx,
+		http.MethodGet,
+		fmt.Sprintf("https://%s/_cluster/health", dsn),
+		nil,
+	)
+	if err != nil {
+		return fmt.Errorf("failed to create Elasticsearch health check request: %w", err)
+	}
+
+	req.SetBasicAuth("elastic", password)
+
+	resp, err := client.Do(req)
+	if err != nil {
+		return fmt.Errorf("failed to send Elasticsearch health check request: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("unexpected status code from Elasticsearch health check: %d", resp.StatusCode)
+	}
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return fmt.Errorf("failed to read Elasticsearch health check response: %w", err)
+	}
+
+	healthResp := struct {
+		Status string `json:"status"`
+	}{}
+
+	if err := json.Unmarshal(body, &healthResp); err != nil {
+		return fmt.Errorf("failed to parse Elasticsearch health check response: %w", err)
+	}
+
+	if healthResp.Status != "green" {
+		return fmt.Errorf("elasticsearch cluster status is not green: %s", healthResp.Status)
+	}
+	return nil
+}

checks/elasticsearch/check_test.go

@@ -0,0 +1,36 @@
+package elasticsearch
+
+import (
+	"context"
+	"os"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+const esDSNEnv = "HEALTH_GO_ES_DSN"
+
+func TestNew(t *testing.T) {
+	t.Parallel()
+
+	check := New(getConfig(t))
+
+	err := check(context.Background())
+	require.NoError(t, err)
+}
+
+func getConfig(t *testing.T) Config {
+	t.Helper()
+
+	elasticSearchDSN, ok := os.LookupEnv(esDSNEnv)
+	require.True(t, ok, "HEALTH_GO_ES_DSN environment variable not set")
+
+	// "docker-compose port <service> <port>" returns 0.0.0.0:XXXX locally, change it to local port
+	elasticSearchDSN = strings.Replace(elasticSearchDSN, "0.0.0.0:", "127.0.0.1:", 1)
+
+	return Config{
+		DSN:      elasticSearchDSN,
+		Password: "test", // Set in docker-compose.yml
+	}
+}

docker-compose.yml

@@ -131,4 +131,17 @@ services:
     image: nats:2.9.11
     command: "-js -sd /data"
     ports:
-      - "4222:4222"
+      - "4222:4222"
+
+  elasticsearch:
+    image: docker.elastic.co/elasticsearch/elasticsearch:8.8.2
+    ports:
+      - "9200:9200"
+    environment:
+      - discovery.type=single-node
+      - ELASTIC_PASSWORD=test
+    healthcheck:
+      test: ["CMD-SHELL", "curl -k -u elastic:test https://localhost:9200 >/dev/null || exit 1"]
+      interval: 15s
+      timeout: 5s
+      retries: 5