Add session health check management and tests

Author: srahul3

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 1.21.0 (March 17, 2025)
+* Enhancement: Added support for Consul Session to update the state of a Health Check, allowing for more dynamic and responsive health monitoring within the Consul ecosystem. This feature enables sessions to directly influence health check statuses, improving the overall reliability and accuracy of service health assessments.
+
 ## 1.20.3 (February 13, 2025)
 
 SECURITY:

agent/consul/state/session.go

@@ -5,6 +5,7 @@ package state
 
 import (
 	"fmt"
+	"github.com/hashicorp/consul/api"
 	"reflect"
 	"strings"
 	"time"
@@ -448,5 +449,28 @@ func (s *Store) deleteSessionTxn(tx WriteTxn, idx uint64, sessionID string, entM
 		}
 	}
 
+	// session invalidating the health-checks
+	return s.markSessionCheckCritical(tx, idx, session)
+}
+
+// markSessionCheckCritical The method marks the health-checks associated with the session as critical
+func (s *Store) markSessionCheckCritical(tx WriteTxn, idx uint64, session *structs.Session) error {
+	// Find all checks for the given Node
+	iter, err := tx.Get(tableChecks, indexNode, Query{Value: session.Node, EnterpriseMeta: session.EnterpriseMeta})
+	if err != nil {
+		return fmt.Errorf("failed check lookup: %s", err)
+	}
+
+	for check := iter.Next(); check != nil; check = iter.Next() {
+		if hc := check.(*structs.HealthCheck); hc.Type == "session" && hc.Definition.SessionName == session.Name {
+			updatedCheck := hc.Clone()
+			updatedCheck.Status = api.HealthCritical
+			updatedCheck.Output = fmt.Sprintf("Session '%s' is invalid", session.Name)
+
+			if err := s.ensureCheckTxn(tx, idx, true, updatedCheck); err != nil {
+				return err
+			}
+		}
+	}
 	return nil
 }

agent/consul/state/session_test.go

@@ -961,3 +961,74 @@ func TestStateStore_Session_Invalidate_PreparedQuery_Delete(t *testing.T) {
 		t.Fatalf("bad: %v", q2)
 	}
 }
+
+func TestHealthCheck_SessionDestroy(t *testing.T) {
+	s := testStateStore(t)
+
+	var check *structs.HealthCheck
+	// setup node
+	testRegisterNode(t, s, 1, "foo-node")
+	testRegisterCheckCustom(t, s, 1, "foo", func(chk *structs.HealthCheck) {
+		chk.Node = "foo-node"
+		chk.Type = "session"
+		chk.Status = api.HealthPassing
+		chk.Definition = structs.HealthCheckDefinition{
+			SessionName: "test-session",
+		}
+		check = chk
+	})
+
+	// Ensure the index was not updated if nothing was destroyed.
+	if idx := s.maxIndex("sessions"); idx != 0 {
+		t.Fatalf("bad index: %d", idx)
+	}
+
+	// Register a new session
+	sess := &structs.Session{
+		ID:   testUUID(),
+		Node: "foo-node",
+		Name: "test-session",
+	}
+	if err := s.SessionCreate(2, sess); err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	_, hc, err := s.ChecksInState(nil, api.HealthAny, structs.DefaultEnterpriseMetaInPartition(""), structs.DefaultPeerKeyword)
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+	// iterate over checks and find the right one
+	found := false
+	for _, c := range hc {
+		if c.CheckID == check.CheckID && c.Status == api.HealthPassing {
+			found = true
+			break
+		}
+	}
+
+	if !found {
+		t.Fatalf("check is expected to be passing")
+	}
+
+	// Destroy the session.
+	if err := s.SessionDestroy(3, sess.ID, nil); err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	_, hc, err = s.ChecksInState(nil, api.HealthAny, structs.DefaultEnterpriseMetaInPartition(""), structs.DefaultPeerKeyword)
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+	// iterate over checks and find the right one
+	found = false
+	for _, c := range hc {
+		if c.CheckID == check.CheckID && c.Status == api.HealthCritical {
+			found = true
+			break
+		}
+	}
+
+	if !found {
+		t.Fatalf("check is expected to be critical")
+	}
+}

agent/consul/state/state_store_test.go

@@ -308,6 +308,29 @@ func testRegisterCheckWithPartition(t *testing.T, s *Store, idx uint64,
 	}
 }
 
+func testRegisterCheckCustom(t *testing.T, s *Store, idx uint64, checkID types.CheckID, update func(chk *structs.HealthCheck)) {
+	chk := &structs.HealthCheck{
+		CheckID:        checkID,
+		EnterpriseMeta: *structs.DefaultEnterpriseMetaInPartition(""),
+	}
+
+	update(chk)
+
+	if err := s.EnsureCheck(idx, chk); err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	tx := s.db.Txn(false)
+	defer tx.Abort()
+	c, err := tx.First(tableChecks, indexID, NodeCheckQuery{Node: chk.Node, CheckID: string(checkID), EnterpriseMeta: chk.EnterpriseMeta})
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+	if result, ok := c.(*structs.HealthCheck); !ok || result.CheckID != checkID {
+		t.Fatalf("bad check: %#v", result)
+	}
+}
+
 func testRegisterSidecarProxy(t *testing.T, s *Store, idx uint64, nodeID string, targetServiceID string) {
 	testRegisterSidecarProxyOpts(t, s, idx, nodeID, targetServiceID)
 }

agent/structs/structs.go

@@ -1912,6 +1912,7 @@ type HealthCheckDefinition struct {
 	GRPCUseTLS                     bool                `json:",omitempty"`
 	AliasNode                      string              `json:",omitempty"`
 	AliasService                   string              `json:",omitempty"`
+	SessionName                    string              `json:",omitempty"`
 	TTL                            time.Duration       `json:",omitempty"`
 }
 

api/health.go

@@ -75,6 +75,8 @@ type HealthCheckDefinition struct {
 	IntervalDuration                       time.Duration `json:"-"`
 	TimeoutDuration                        time.Duration `json:"-"`
 	DeregisterCriticalServiceAfterDuration time.Duration `json:"-"`
+	// when parent Type is `session`, and if this session is destroyed, the check will be marked as critical
+	SessionName string `json:",omitempty"`
 
 	// DEPRECATED in Consul 1.4.1. Use the above time.Duration fields instead.
 	Interval                       ReadableDuration