ITWorx-Pulse-Public/internal/agentsource/application.go

package agentsource

import (
	"context"
	"errors"
	"strings"
	"time"

	"github.com/itworx/pulse/internal/application"
	"github.com/itworx/pulse/internal/container"
	"github.com/itworx/pulse/internal/service"
)

// ApplicationProvider derives application health from two existing surfaces rather than
// from a capability of its own: the container inventory the agent records, and the
// service probe results the API already stores. Containers are grouped into applications
// by their Compose project (falling back to the container name for a standalone
// container), and a probe result for a service with the same name refines the component
// status of the matching container.
//
// Fresh container evidence is required. Service probes refine matching
// components when configured; their absence does not erase container-runtime
// availability, because those are separate claims and surfaces.
type ApplicationProvider struct {
	Containers container.Provider
	Services   service.Provider
	// MaxApplications bounds the number of groups built before the payload is refused.
	// Zero takes DefaultMaxApplications.
	MaxApplications int
	Now             func() time.Time
}

var _ application.Provider = ApplicationProvider{}

// DefaultMaxApplications matches the bound application.BuildSnapshot enforces, so an
// oversized inventory is reported as Unknown instead of failing the request.
const DefaultMaxApplications = 150

func (p ApplicationProvider) Snapshot(ctx context.Context) (application.Snapshot, error) {
	if err := ctx.Err(); err != nil {
		return application.Snapshot{}, err
	}
	now := clockNow(p.Now)
	unknown := func(reason string) application.Snapshot {
		return application.UnknownSnapshot(now, applicationSources, agentSourceType, reason)
	}
	if p.Containers == nil {
		return unknown(ReasonUnavailable), nil
	}
	containers, err := p.Containers.Snapshot(ctx)
	if err != nil {
		if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
			return application.Snapshot{}, err
		}
		return unknown(ReasonUnavailable), nil
	}
	if reason := unusableContainerReason(containers); reason != "" {
		return unknown(reason), nil
	}
	// Service probes refine application health when configured, but are not an
	// identity prerequisite. Fresh container state is complete evidence for
	// runtime availability; an empty/disabled probe module must not erase every
	// discovered application from the product.
	services := service.Snapshot{ContractVersion: service.ContractVersion, ObservedAt: containers.Source.ObservedAt}
	if p.Services != nil {
		observed, serviceErr := p.Services.Snapshot(ctx)
		if serviceErr != nil {
			if errors.Is(serviceErr, context.Canceled) || errors.Is(serviceErr, context.DeadlineExceeded) {
				return application.Snapshot{}, serviceErr
			}
		} else if strings.TrimSpace(observed.Reason) == "" {
			services = observed
		}
	}
	maximum := p.MaxApplications
	if maximum <= 0 {
		maximum = DefaultMaxApplications
	}
	discovered := groupApplications(containers, services, maximum)
	if discovered == nil {
		return unknown(ReasonInvalid), nil
	}
	source := application.Source{
		ID:         applicationSources,
		Type:       agentSourceType,
		ObservedAt: containers.Source.ObservedAt,
		ReceivedAt: containers.Source.ReceivedAt,
	}
	snapshot, err := application.BuildSnapshot(source, discovered, nil, now)
	if err != nil {
		return unknown(ReasonInvalid), nil
	}
	return snapshot, nil
}

// unusableContainerReason maps the container source state onto an application reason code.
// The container provider has already applied the freshness rule, so an Unknown container
// source here means the same thing for applications.
func unusableContainerReason(snapshot container.Snapshot) string {
	if snapshot.Source.State == "" || snapshot.Source.State == "unknown" {
		switch snapshot.Source.Reason {
		case ReasonStale, "stale_source":
			return ReasonStale
		case ReasonInvalid:
			return ReasonInvalid
		default:
			return ReasonUnavailable
		}
	}
	return ""
}

// groupApplications builds one discovered application per Compose project. It returns nil
// when the inventory exceeds the configured bound.
func groupApplications(containers container.Snapshot, services service.Snapshot, maximum int) []application.DiscoveredApplication {
	states := make(map[string]application.State, len(services.Services))
	for _, item := range services.Services {
		states[normalizeKey(item.Name)] = serviceState(item.State)
	}
	// application.evaluateComponent normalizes an empty service state to Unknown, so a
	// component with no probe would drag its application to Unknown even though the
	// container telemetry is fresh. There is no probe constraint on such a component, and
	// Healthy is the neutral element of the domain's aggregation: it leaves the component
	// status equal to the container status instead of inventing an unknown.
	stateFor := func(name string) application.State {
		if state, ok := states[normalizeKey(name)]; ok {
			return state
		}
		return application.StateHealthy
	}
	order := make([]string, 0, len(containers.Containers))
	groups := make(map[string]*application.DiscoveredApplication, len(containers.Containers))
	for _, item := range containers.Containers {
		key := strings.TrimSpace(item.Project)
		if key == "" {
			key = strings.TrimPrefix(strings.TrimSpace(item.Name), "/")
		}
		if key == "" {
			continue
		}
		group, seen := groups[key]
		if !seen {
			if len(order) >= maximum {
				return nil
			}
			group = &application.DiscoveredApplication{ID: application.StableApplicationID(application.SourceID, key), Name: key}
			groups[key] = group
			order = append(order, key)
		}
		group.Components = append(group.Components, application.ComponentInput{
			ID:             item.ID,
			Name:           strings.TrimPrefix(item.Name, "/"),
			Kind:           "container",
			ContainerState: containerState(item),
			ServiceState:   stateFor(item.Name),
			// Every discovered component is critical until an operator overrides it;
			// treating an unmapped component as optional would hide real failures.
			Critical: true,
		})
	}
	discovered := make([]application.DiscoveredApplication, 0, len(order))
	for _, key := range order {
		discovered = append(discovered, *groups[key])
	}
	return discovered
}

// containerState maps a normalized container onto the application state vocabulary. A
// container the operator stopped on purpose is not a failure; anything the collector
// could not classify is Unknown rather than Healthy.
func containerState(item container.Container) application.State {
	switch strings.ToLower(strings.TrimSpace(item.State)) {
	case "running":
		switch strings.ToLower(strings.TrimSpace(item.Health)) {
		case "healthy":
			return application.StateHealthy
		case "unhealthy":
			return application.StateDegraded
		case "starting", "unknown", "":
			return application.StateUnknown
		default:
			return application.StateUnknown
		}
	case "restarting", "paused", "removing":
		return application.StateDegraded
	case "exited", "dead", "stopped":
		if item.IntentionalStop {
			return application.StateUnknown
		}
		return application.StateDown
	default:
		return application.StateUnknown
	}
}

// serviceState maps a probe verdict onto the application state vocabulary.
func serviceState(state string) application.State {
	switch strings.ToLower(strings.TrimSpace(state)) {
	case service.StateUp:
		return application.StateHealthy
	case service.StateDegraded:
		return application.StateDegraded
	case service.StateDown:
		return application.StateDown
	default:
		return application.StateUnknown
	}
}

func normalizeKey(value string) string {
	return strings.ToLower(strings.TrimPrefix(strings.TrimSpace(value), "/"))
}