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

package agentsource

import (
	"context"
	"time"

	"github.com/itworx/pulse/internal/agentstore"
	"github.com/itworx/pulse/internal/application"
	"github.com/itworx/pulse/internal/array"
	"github.com/itworx/pulse/internal/container"
	"github.com/itworx/pulse/internal/disk"
	"github.com/itworx/pulse/internal/host"
	"github.com/itworx/pulse/internal/pool"
	"github.com/itworx/pulse/internal/process"
	"github.com/itworx/pulse/internal/share"
)

// Source identifiers and types reported to the API. They match the defaults each domain
// Adapter already applies, so a snapshot served from the agent store is indistinguishable
// in shape from one served by any other adapter. The type records where the observation
// originates: the host capabilities are read by the agent itself, the storage
// capabilities are read by the agent from Unraid.
const (
	hostSourceID       = "host"
	processSourceID    = "process"
	containerSourceID  = "container"
	arraySourceID      = "array"
	diskSourceID       = "disks"
	poolSourceID       = "pools"
	shareSourceID      = "shares"
	agentSourceType    = "agent"
	unraidSourceType   = "unraid"
	applicationSources = application.SourceID
)

// HostProvider serves host telemetry recorded by the agent.
type HostProvider struct {
	Reader  agentstore.Reader
	Windows Windows
	Limits  host.Limits
	Policy  host.Policy
	// Now overrides the clock in tests; production leaves it nil.
	Now func() time.Time
}

var _ host.Provider = HostProvider{}

func (p HostProvider) Snapshot(ctx context.Context) (host.Snapshot, error) {
	now := clockNow(p.Now)
	return resolve(ctx, p.Reader, agentstore.CapabilityHost, p.Windows.For(agentstore.CapabilityHost), now,
		func(at time.Time, reason string) host.Snapshot {
			return host.UnknownSnapshot(at, hostSourceID, agentSourceType, reason)
		},
		func(raw host.RawSnapshot, at time.Time) (host.Snapshot, error) {
			return host.Adapter{Source: staticRaw[host.RawSnapshot]{raw}, Limits: p.Limits, Policy: p.Policy, Now: fixedClock(at)}.Snapshot(ctx)
		})
}

// ProcessProvider serves the process inventory recorded by the agent.
type ProcessProvider struct {
	Reader  agentstore.Reader
	Windows Windows
	Limits  process.Limits
	Now     func() time.Time
}

func (p ProcessProvider) Snapshot(ctx context.Context) (process.Snapshot, error) {
	now := clockNow(p.Now)
	return resolve(ctx, p.Reader, agentstore.CapabilityProcesses, p.Windows.For(agentstore.CapabilityProcesses), now,
		func(at time.Time, reason string) process.Snapshot {
			return process.UnknownSnapshot(at, processSourceID, agentSourceType, reason)
		},
		func(raw process.RawSnapshot, at time.Time) (process.Snapshot, error) {
			return process.Adapter{Source: staticRaw[process.RawSnapshot]{raw}, Limits: p.Limits, Now: fixedClock(at)}.Snapshot(ctx)
		})
}

// ContainerProvider serves the container inventory recorded by the agent.
type ContainerProvider struct {
	Reader  agentstore.Reader
	Windows Windows
	Limits  container.Limits
	Now     func() time.Time
}

var _ container.Provider = ContainerProvider{}

func (p ContainerProvider) Snapshot(ctx context.Context) (container.Snapshot, error) {
	now := clockNow(p.Now)
	return resolve(ctx, p.Reader, agentstore.CapabilityContainers, p.Windows.For(agentstore.CapabilityContainers), now,
		func(at time.Time, reason string) container.Snapshot {
			return container.UnknownSnapshot(at, containerSourceID, agentSourceType, reason)
		},
		func(raw container.RawSnapshot, at time.Time) (container.Snapshot, error) {
			return container.Adapter{Source: staticRaw[container.RawSnapshot]{raw}, Limits: p.Limits, Now: fixedClock(at)}.Snapshot(ctx)
		})
}

// ArrayProvider serves Unraid array state recorded by the agent.
type ArrayProvider struct {
	Reader  agentstore.Reader
	Windows Windows
	Limits  array.Limits
	Policy  array.Policy
	Now     func() time.Time
}

var _ array.Provider = ArrayProvider{}

func (p ArrayProvider) Snapshot(ctx context.Context) (array.Snapshot, error) {
	now := clockNow(p.Now)
	return resolve(ctx, p.Reader, agentstore.CapabilityArray, p.Windows.For(agentstore.CapabilityArray), now,
		func(at time.Time, reason string) array.Snapshot {
			return array.UnknownSnapshot(at, arraySourceID, unraidSourceType, reason)
		},
		func(raw array.RawSnapshot, at time.Time) (array.Snapshot, error) {
			return array.Adapter{Source: staticRaw[array.RawSnapshot]{raw}, Limits: p.Limits, Policy: p.Policy, Now: fixedClock(at)}.Snapshot(ctx)
		})
}

// DiskProvider serves disk inventory, SMART and performance data recorded by the agent.
type DiskProvider struct {
	Reader  agentstore.Reader
	Windows Windows
	Limits  disk.Limits
	Policy  disk.Policy
	Now     func() time.Time
}

var _ disk.Provider = DiskProvider{}

func (p DiskProvider) Snapshot(ctx context.Context) (disk.Snapshot, error) {
	now := clockNow(p.Now)
	return resolve(ctx, p.Reader, agentstore.CapabilityDisks, p.Windows.For(agentstore.CapabilityDisks), now,
		func(at time.Time, reason string) disk.Snapshot {
			return disk.UnknownSnapshot(at, diskSourceID, unraidSourceType, reason)
		},
		func(raw disk.RawSnapshot, at time.Time) (disk.Snapshot, error) {
			return disk.Adapter{Source: staticRaw[disk.RawSnapshot]{raw}, Limits: p.Limits, Policy: p.Policy, Now: fixedClock(at)}.Snapshot(ctx)
		})
}

// PoolProvider serves cache and named pool state recorded by the agent.
type PoolProvider struct {
	Reader  agentstore.Reader
	Windows Windows
	Limits  pool.Limits
	Policy  pool.Policy
	Now     func() time.Time
}

var _ pool.Provider = PoolProvider{}

func (p PoolProvider) Snapshot(ctx context.Context) (pool.Snapshot, error) {
	now := clockNow(p.Now)
	return resolve(ctx, p.Reader, agentstore.CapabilityPools, p.Windows.For(agentstore.CapabilityPools), now,
		func(at time.Time, reason string) pool.Snapshot {
			return pool.UnknownSnapshot(at, poolSourceID, unraidSourceType, reason)
		},
		func(raw pool.RawSnapshot, at time.Time) (pool.Snapshot, error) {
			return pool.Adapter{Source: staticRaw[pool.RawSnapshot]{raw}, Limits: p.Limits, Policy: p.Policy, Now: fixedClock(at)}.Snapshot(ctx)
		})
}

// ShareProvider serves user share usage recorded by the agent.
type ShareProvider struct {
	Reader  agentstore.Reader
	Windows Windows
	Limits  share.Limits
	Policy  share.Policy
	Now     func() time.Time
}

var _ share.Provider = ShareProvider{}

func (p ShareProvider) Snapshot(ctx context.Context) (share.Snapshot, error) {
	now := clockNow(p.Now)
	return resolve(ctx, p.Reader, agentstore.CapabilityShares, p.Windows.For(agentstore.CapabilityShares), now,
		func(at time.Time, reason string) share.Snapshot {
			return share.UnknownSnapshot(at, shareSourceID, unraidSourceType, reason)
		},
		func(raw share.RawSnapshot, at time.Time) (share.Snapshot, error) {
			return share.Adapter{Source: staticRaw[share.RawSnapshot]{raw}, Limits: p.Limits, Policy: p.Policy, Now: fixedClock(at)}.Snapshot(ctx)
		})
}