(airavata-custos) 10/18: Add sign-in status probe to the identity provisioner

[la...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

lahirujayathilake pushed a commit to branch pearc26
in repository https://gitbox.apache.org/repos/asf/airavata-custos.git

commit 9a439b7cc2615443fbf101f1f9ff0f92c0dd8d87
Author: lahiruj <la...@gmail.com>
AuthorDate: Fri Jul 24 17:41:10 2026 -0400

    Add sign-in status probe to the identity provisioner
---
 .../internal/subscribers/signin_probe.go           | 174 ++++++++++++++++
 .../internal/subscribers/signin_probe_test.go      | 229 +++++++++++++++++++++
 .../Identity-Provisioner/pkg/comanage/loader.go    |  19 +-
 3 files changed, 421 insertions(+), 1 deletion(-)

diff --git a/connectors/COmanage/Identity-Provisioner/internal/subscribers/signin_probe.go b/connectors/COmanage/Identity-Provisioner/internal/subscribers/signin_probe.go
new file mode 100644
index 000000000..4bc73676b
--- /dev/null
+++ b/connectors/COmanage/Identity-Provisioner/internal/subscribers/signin_probe.go
@@ -0,0 +1,174 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package subscribers
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"log/slog"
+	"time"
+
+	"github.com/apache/airavata-custos/connectors/COmanage/Identity-Provisioner/internal/client"
+	"github.com/apache/airavata-custos/pkg/models"
+	"github.com/apache/airavata-custos/pkg/service"
+)
+
+// DefaultSignInCheckInterval is how often the sign-in health probe runs when
+// the connector config does not set signin_check_interval. The registry is an
+// external hosted service and sign-in state changes rarely, so this stays far
+// slower than the cluster-side probe.
+const DefaultSignInCheckInterval = 10 * time.Minute
+
+const signInProbeTimeout = 2 * time.Minute
+
+// Matches the source the provisioner stores person ids under.
+const registryIdentitySource = "comanage"
+
+// registryReader is the one read the probe performs against the registry.
+type registryReader interface {
+	GetPersonComposite(identifier string) (json.RawMessage, error)
+}
+
+// SignInProbe reports sign-in health for every active member of this
+// connector's cluster: the account must be provisioned and its person record
+// must still resolve in the registry. Read-only on both sides; provisioning
+// stays with the orchestrator.
+//
+// A person whose record resolves but has lost its username identifier is not
+// detected. Catching that needs a composite parse; add it if it ever happens.
+type SignInProbe struct {
+	registry  registryReader
+	core      service.CoreService
+	clusterID string
+}
+
+func NewSignInProbe(registry registryReader, core service.CoreService, clusterID string) *SignInProbe {
+	return &SignInProbe{registry: registry, core: core, clusterID: clusterID}
+}
+
+// Start runs the probe until ctx is cancelled.
+func (p *SignInProbe) Start(ctx context.Context, interval time.Duration) {
+	if interval <= 0 {
+		interval = DefaultSignInCheckInterval
+	}
+	slog.Info("Starting sign-in access status probe", "interval", interval)
+
+	p.probeOnce(ctx)
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			slog.Info("Stopping sign-in access status probe", "reason", ctx.Err())
+			return
+		case <-ticker.C:
+			p.probeOnce(ctx)
+		}
+	}
+}
+
+func (p *SignInProbe) probeOnce(ctx context.Context) {
+	defer func() {
+		if r := recover(); r != nil {
+			slog.Error("Sign-in probe: pass panicked", "panic", r)
+		}
+	}()
+	probeCtx, cancel := context.WithTimeout(ctx, signInProbeTimeout)
+	defer cancel()
+
+	clusterUsers, err := p.core.ListComputeClusterUsersByCluster(probeCtx, p.clusterID)
+	if err != nil {
+		slog.Error("Sign-in probe: failed to list cluster users", "cluster_id", p.clusterID, "error", err)
+		return
+	}
+
+	// One registry lookup per user per pass, shared across their memberships.
+	type verdict struct {
+		ok     bool
+		detail string
+	}
+	verdicts := make(map[string]verdict, len(clusterUsers))
+
+	for _, csu := range clusterUsers {
+		memberships, err := p.core.ListAllocationsForUser(probeCtx, csu.UserID)
+		if err != nil {
+			slog.Error("Sign-in probe: failed to list memberships", "user_id", csu.UserID, "error", err)
+			continue
+		}
+		for _, membership := range memberships {
+			if membership.MembershipStatus != models.ACTIVE {
+				continue
+			}
+			alloc, err := p.core.GetComputeAllocation(probeCtx, membership.ComputeAllocationID)
+			if err != nil {
+				slog.Warn("Sign-in probe: could not resolve allocation",
+					"membership_id", membership.ID, "error", err)
+				continue
+			}
+			if alloc.ComputeClusterID != p.clusterID || alloc.Status != models.ACTIVE {
+				continue
+			}
+			v, seen := verdicts[csu.UserID]
+			if !seen {
+				var skip bool
+				v.ok, v.detail, skip = p.verify(probeCtx, csu)
+				if skip {
+					continue
+				}
+				verdicts[csu.UserID] = v
+			}
+			err = p.core.RecordAccessCheckResult(probeCtx, membership.ComputeAllocationID, csu.UserID, models.AccessCheckSignIn, v.ok, v.detail)
+			if err != nil {
+				slog.Error("Sign-in probe: failed to record result",
+					"membership_id", membership.ID, "error", err)
+			}
+		}
+	}
+}
+
+// verify decides one user's sign-in health without writing anything. skip
+// means the truth is unknowable right now (a core lookup failed), which must
+// not be recorded as unhealthy.
+func (p *SignInProbe) verify(ctx context.Context, csu models.ComputeClusterUser) (ok bool, detail string, skip bool) {
+	if csu.ProvisionedAt == nil {
+		return false, "account not provisioned yet", false
+	}
+	idents, err := p.core.ListUserIdentitiesForUser(ctx, csu.UserID)
+	if err != nil {
+		slog.Warn("Sign-in probe: could not list identities", "user_id", csu.UserID, "error", err)
+		return false, "", true
+	}
+	personID := ""
+	for _, id := range idents {
+		if id.Source == registryIdentitySource && id.ExternalID != "" {
+			personID = id.ExternalID
+			break
+		}
+	}
+	if personID == "" {
+		return false, "registry record not linked", false
+	}
+	if _, err := p.registry.GetPersonComposite(personID); err != nil {
+		if errors.Is(err, client.ErrNotFound) {
+			return false, "registry record missing", false
+		}
+		return false, "registry unreachable", false
+	}
+	return true, "", false
+}
diff --git a/connectors/COmanage/Identity-Provisioner/internal/subscribers/signin_probe_test.go b/connectors/COmanage/Identity-Provisioner/internal/subscribers/signin_probe_test.go
new file mode 100644
index 000000000..3dae63dfa
--- /dev/null
+++ b/connectors/COmanage/Identity-Provisioner/internal/subscribers/signin_probe_test.go
@@ -0,0 +1,229 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package subscribers
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/apache/airavata-custos/connectors/COmanage/Identity-Provisioner/internal/client"
+	"github.com/apache/airavata-custos/pkg/models"
+	"github.com/apache/airavata-custos/pkg/service"
+)
+
+type fakeRegistry struct {
+	err   error
+	calls int
+}
+
+func (f *fakeRegistry) GetPersonComposite(string) (json.RawMessage, error) {
+	f.calls++
+	if f.err != nil {
+		return nil, f.err
+	}
+	return json.RawMessage(`{}`), nil
+}
+
+type signInRecorded struct {
+	allocationID string
+	userID       string
+	checkType    models.AccessCheckType
+	ok           bool
+	detail       string
+}
+
+type signInMockOpts struct {
+	provisioned  bool
+	linked       bool
+	memberships  []models.ComputeAllocationMembership
+	allocCluster string
+}
+
+func signInCoreMock(o signInMockOpts, results *[]signInRecorded) *service.CoreServiceMock {
+	provisionedAt := (*time.Time)(nil)
+	if o.provisioned {
+		t := time.Now().Add(-time.Hour)
+		provisionedAt = &t
+	}
+	memberships := o.memberships
+	if memberships == nil {
+		memberships = []models.ComputeAllocationMembership{
+			{ID: "mem-1", ComputeAllocationID: "alloc-1", UserID: "user-1", MembershipStatus: models.ACTIVE},
+		}
+	}
+	allocCluster := o.allocCluster
+	if allocCluster == "" {
+		allocCluster = "cluster-1"
+	}
+	return &service.CoreServiceMock{
+		ListComputeClusterUsersByClusterFunc: func(ctx context.Context, clusterID string) ([]models.ComputeClusterUser, error) {
+			return []models.ComputeClusterUser{
+				{ID: "csu-1", ComputeClusterID: clusterID, UserID: "user-1", LocalUsername: "testuser", ProvisionedAt: provisionedAt},
+			}, nil
+		},
+		ListAllocationsForUserFunc: func(ctx context.Context, userID string) ([]models.ComputeAllocationMembership, error) {
+			return memberships, nil
+		},
+		GetComputeAllocationFunc: func(ctx context.Context, id string) (*models.ComputeAllocation, error) {
+			return &models.ComputeAllocation{ID: id, Name: "alloc", ComputeClusterID: allocCluster, Status: models.ACTIVE}, nil
+		},
+		ListUserIdentitiesForUserFunc: func(ctx context.Context, userID string) ([]models.UserIdentity, error) {
+			if !o.linked {
+				return nil, nil
+			}
+			return []models.UserIdentity{{UserID: userID, Source: "comanage", ExternalID: "person-9"}}, nil
+		},
+		RecordAccessCheckResultFunc: func(ctx context.Context, allocationID, userID string, checkType models.AccessCheckType, ok bool, detail string) error {
+			*results = append(*results, signInRecorded{allocationID, userID, checkType, ok, detail})
+			return nil
+		},
+	}
+}
+
+func TestSignInProbe_ProvisionedAndResolvableRecordsOK(t *testing.T) {
+	var results []signInRecorded
+	reg := &fakeRegistry{}
+	NewSignInProbe(reg, signInCoreMock(signInMockOpts{provisioned: true, linked: true}, &results), "cluster-1").probeOnce(context.Background())
+
+	if len(results) != 1 {
+		t.Fatalf("results: got %+v, want 1", results)
+	}
+	r := results[0]
+	if !r.ok || r.checkType != models.AccessCheckSignIn || r.allocationID != "alloc-1" {
+		t.Errorf("recorded: %+v, want ok SIGN_IN for alloc-1", r)
+	}
+}
+
+func TestSignInProbe_UnprovisionedRecordsNotProvisioned(t *testing.T) {
+	var results []signInRecorded
+	reg := &fakeRegistry{}
+	NewSignInProbe(reg, signInCoreMock(signInMockOpts{provisioned: false, linked: true}, &results), "cluster-1").probeOnce(context.Background())
+
+	if len(results) != 1 || results[0].ok || results[0].detail != "account not provisioned yet" {
+		t.Fatalf("recorded: %+v, want failure 'account not provisioned yet'", results)
+	}
+	if reg.calls != 0 {
+		t.Errorf("registry consulted for an unprovisioned account: %d calls", reg.calls)
+	}
+}
+
+func TestSignInProbe_RegistryOutcomes(t *testing.T) {
+	cases := []struct {
+		name   string
+		err    error
+		detail string
+	}{
+		{"missing person", client.ErrNotFound, "registry record missing"},
+		{"registry down", errors.New("http 503"), "registry unreachable"},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			var results []signInRecorded
+			reg := &fakeRegistry{err: tc.err}
+			NewSignInProbe(reg, signInCoreMock(signInMockOpts{provisioned: true, linked: true}, &results), "cluster-1").probeOnce(context.Background())
+			if len(results) != 1 || results[0].ok || results[0].detail != tc.detail {
+				t.Fatalf("recorded: %+v, want failure %q", results, tc.detail)
+			}
+		})
+	}
+}
+
+func TestSignInProbe_UnlinkedRecordsNotLinked(t *testing.T) {
+	var results []signInRecorded
+	reg := &fakeRegistry{}
+	NewSignInProbe(reg, signInCoreMock(signInMockOpts{provisioned: true, linked: false}, &results), "cluster-1").probeOnce(context.Background())
+
+	if len(results) != 1 || results[0].ok || results[0].detail != "registry record not linked" {
+		t.Fatalf("recorded: %+v, want failure 'registry record not linked'", results)
+	}
+	if reg.calls != 0 {
+		t.Errorf("registry consulted without a stored person id: %d calls", reg.calls)
+	}
+}
+
+func TestSignInProbe_OneLookupSharedAcrossMemberships(t *testing.T) {
+	var results []signInRecorded
+	reg := &fakeRegistry{}
+	mock := signInCoreMock(signInMockOpts{
+		provisioned: true, linked: true,
+		memberships: []models.ComputeAllocationMembership{
+			{ID: "mem-1", ComputeAllocationID: "alloc-1", UserID: "user-1", MembershipStatus: models.ACTIVE},
+			{ID: "mem-2", ComputeAllocationID: "alloc-2", UserID: "user-1", MembershipStatus: models.ACTIVE},
+			{ID: "mem-3", ComputeAllocationID: "alloc-3", UserID: "user-1", MembershipStatus: models.INACTIVE},
+		},
+	}, &results)
+	NewSignInProbe(reg, mock, "cluster-1").probeOnce(context.Background())
+
+	if len(results) != 2 {
+		t.Fatalf("results: got %+v, want 2 (both ACTIVE memberships)", results)
+	}
+	if reg.calls != 1 {
+		t.Errorf("registry lookups: got %d, want 1 shared", reg.calls)
+	}
+}
+
+func TestSignInProbe_SkipsOtherClustersAllocations(t *testing.T) {
+	var results []signInRecorded
+	reg := &fakeRegistry{}
+	mock := signInCoreMock(signInMockOpts{provisioned: true, linked: true, allocCluster: "cluster-other"}, &results)
+	NewSignInProbe(reg, mock, "cluster-1").probeOnce(context.Background())
+
+	if len(results) != 0 {
+		t.Fatalf("recorded for another cluster's allocation: %+v", results)
+	}
+}
+
+func TestSignInProbe_CoreLookupFailureRecordsNothing(t *testing.T) {
+	var results []signInRecorded
+	reg := &fakeRegistry{}
+	mock := signInCoreMock(signInMockOpts{provisioned: true, linked: true}, &results)
+	mock.ListUserIdentitiesForUserFunc = func(ctx context.Context, userID string) ([]models.UserIdentity, error) {
+		return nil, errors.New("db down")
+	}
+	NewSignInProbe(reg, mock, "cluster-1").probeOnce(context.Background())
+
+	if len(results) != 0 {
+		t.Fatalf("core failure recorded %+v, want nothing (unknowable, not unhealthy)", results)
+	}
+}
+
+func TestSignInProbe_SkipsInactiveAllocation(t *testing.T) {
+	var results []signInRecorded
+	reg := &fakeRegistry{}
+	mock := signInCoreMock(signInMockOpts{provisioned: true, linked: true}, &results)
+	mock.GetComputeAllocationFunc = func(ctx context.Context, id string) (*models.ComputeAllocation, error) {
+		return &models.ComputeAllocation{ID: id, Name: "alloc", ComputeClusterID: "cluster-1", Status: models.INACTIVE}, nil
+	}
+	NewSignInProbe(reg, mock, "cluster-1").probeOnce(context.Background())
+
+	if len(results) != 0 {
+		t.Fatalf("inactive allocation recorded %+v, want nothing", results)
+	}
+}
+
+func TestSignInProbe_PassSurvivesPanic(t *testing.T) {
+	var results []signInRecorded
+	mock := signInCoreMock(signInMockOpts{provisioned: true, linked: true}, &results)
+	mock.ListComputeClusterUsersByClusterFunc = func(ctx context.Context, clusterID string) ([]models.ComputeClusterUser, error) {
+		panic("boom")
+	}
+	NewSignInProbe(&fakeRegistry{}, mock, "cluster-1").probeOnce(context.Background())
+}
diff --git a/connectors/COmanage/Identity-Provisioner/pkg/comanage/loader.go b/connectors/COmanage/Identity-Provisioner/pkg/comanage/loader.go
index df6cb0296..b2327266e 100644
--- a/connectors/COmanage/Identity-Provisioner/pkg/comanage/loader.go
+++ b/connectors/COmanage/Identity-Provisioner/pkg/comanage/loader.go
@@ -45,7 +45,7 @@ func init() {
 // LoadConnector wires the subscriber to the event bus. Reads YAML config first
 // and falls back to environment variables. If neither yields a complete
 // config, it logs and returns nil without registering.
-func LoadConnector(_ context.Context, _ *sqlx.DB, eventBus *events.Bus, coreService *service.Service, _ *sync.WaitGroup, _ *identity.Router, connectorConfig *config.ConnectorConfig) error {
+func LoadConnector(ctx context.Context, _ *sqlx.DB, eventBus *events.Bus, coreService *service.Service, wg *sync.WaitGroup, _ *identity.Router, connectorConfig *config.ConnectorConfig) error {
 	cfg, ok := loadConfigFromConnectorConfig(connectorConfig)
 	if !ok {
 		cfg, ok = loadConfigFromEnv()
@@ -58,6 +58,23 @@ func LoadConnector(_ context.Context, _ *sqlx.DB, eventBus *events.Bus, coreServ
 	subscribers.NewClusterUserSubscriber(httpClient, eventBus, coreService, cfg.CustosClusterID).RegisterSubscribers()
 	slog.Info("comanage provisioner: subscriber registered",
 		"registry", cfg.RegistryURL, "co_id", cfg.COID, "cluster_id", cfg.CustosClusterID)
+
+	var signInCheckInterval time.Duration
+	if connectorConfig != nil {
+		if si, ok := connectorConfig.Config["signin_check_interval"].(string); ok && si != "" {
+			if d, err := time.ParseDuration(si); err == nil {
+				signInCheckInterval = d
+			} else {
+				slog.Warn("invalid signin_check_interval, using default", "value", si, "error", err)
+			}
+		}
+	}
+	probe := subscribers.NewSignInProbe(httpClient, coreService, cfg.CustosClusterID)
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		probe.Start(ctx, signInCheckInterval)
+	}()
 	// Custos is the source of truth: CoPerson and UnixClusterAccount records
 	// for users provisioned via this connector must not be edited directly in
 	// COmanage. There is no drift reconciliation; out-of-band edits will be