Help us shape the future of Go! Take the 2026 Go Developer Survey

admin

package
v0.1.0-beta.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 31, 2026 License: Apache-2.0 Imports: 64 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ObjectOperationsTotal counts S3 object operations by type and status.
	ObjectOperationsTotal = promauto.NewCounterVec(prometheus.CounterOpts{
		Namespace: "synaps3",
		Subsystem: "backend",
		Name:      "object_operations_total",
		Help:      "Total S3 object operations by type and status",
	}, []string{"operation", "status"})

	// CacheUsedBytes tracks current cache usage in bytes.
	CacheUsedBytes = promauto.NewGauge(prometheus.GaugeOpts{
		Namespace: "synaps3",
		Subsystem: "cache",
		Name:      "used_bytes",
		Help:      "Current cache usage in bytes",
	})

	// CacheHitsTotal counts cache hits (object found locally).
	CacheHitsTotal = promauto.NewCounter(prometheus.CounterOpts{
		Namespace: "synaps3",
		Subsystem: "cache",
		Name:      "hits_total",
		Help:      "Total cache hits (object found locally)",
	})

	// CacheMissesTotal counts cache misses (object not found locally).
	CacheMissesTotal = promauto.NewCounter(prometheus.CounterOpts{
		Namespace: "synaps3",
		Subsystem: "cache",
		Name:      "misses_total",
		Help:      "Total cache misses (object not found locally)",
	})

	// CacheLRUEvictionPaused reports whether LRU eviction is paused because
	// recent cache access could not be retained safely.
	CacheLRUEvictionPaused = promauto.NewGauge(prometheus.GaugeOpts{
		Namespace: "synaps3",
		Subsystem: "cache",
		Name:      "lru_eviction_paused",
		Help:      "Whether LRU cache eviction is paused because recent cache access could not be retained safely",
	})

	// WorkerTasksProcessed counts tasks processed by worker type and result.
	WorkerTasksProcessed = promauto.NewCounterVec(prometheus.CounterOpts{
		Namespace: "synaps3",
		Subsystem: "worker",
		Name:      "tasks_processed_total",
		Help:      "Total tasks processed by worker type and result",
	}, []string{"worker", "result"})

	// TasksExhaustedTotal counts tasks that entered exhausted status by worker type and task type.
	TasksExhaustedTotal = promauto.NewCounterVec(prometheus.CounterOpts{
		Namespace: "synaps3",
		Subsystem: "worker",
		Name:      "tasks_exhausted_total",
		Help:      "Total tasks that entered exhausted status",
	}, []string{"worker", "task_type"})

	// TaskQueueDepth tracks active task count by type and status.
	TaskQueueDepth = promauto.NewGaugeVec(prometheus.GaugeOpts{
		Namespace: "synaps3",
		Subsystem: "task",
		Name:      "queue_depth",
		Help:      "Number of tasks by type and status",
	}, []string{"type", "status"})

	// WorkerTaskDuration tracks per-task processing duration in seconds.
	WorkerTaskDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
		Namespace: "synaps3",
		Subsystem: "worker",
		Name:      "task_duration_seconds",
		Help:      "Task processing duration in seconds by worker",
		Buckets:   prometheus.ExponentialBuckets(0.1, 2, 12),
	}, []string{"worker"})

	// ObjectStateDistribution tracks object count by state.
	ObjectStateDistribution = promauto.NewGaugeVec(prometheus.GaugeOpts{
		Namespace: "synaps3",
		Subsystem: "object",
		Name:      "state_distribution",
		Help:      "Number of objects by state",
	}, []string{"state"})
)

Functions

This section is empty.

Types

type EventHub 

type EventHub struct {
	// contains filtered or unexported fields
}

func NewEventHub 

func NewEventHub() *EventHub

func (*EventHub) Publish 

func (h *EventHub) Publish(topic string, payload map[string]any)

type EventPublisher 

type EventPublisher interface {
	Publish(topic string, payload map[string]any)
}

type ProviderIdentityResolver 

type ProviderIdentityResolver struct {
	// contains filtered or unexported fields
}

ProviderIdentityResolver asynchronously enriches Registry provider IDs for admin API responses.

func NewProviderIdentityResolver 

func NewProviderIdentityResolver(registry *spregistry.Service, rpcURL string, logger *slog.Logger) *ProviderIdentityResolver

NewProviderIdentityResolver creates a cached provider identity resolver backed by the Registry and Filecoin RPC.

func (*ProviderIdentityResolver) ProviderIdentities 

func (r *ProviderIdentityResolver) ProviderIdentities(providerIDs []idtypes.OnChainID) map[string]*providerIdentityResponse

func (*ProviderIdentityResolver) Run 

func (r *ProviderIdentityResolver) Run(ctx context.Context)

func (*ProviderIdentityResolver) SetProviderIdentityPublisher 

func (r *ProviderIdentityResolver) SetProviderIdentityPublisher(publish func(*providerIdentityResponse))

type ProviderRegistry 

type ProviderRegistry interface {
	ListProviderObservations(ctx context.Context, opts observability.ListOptions) (observability.ProviderObservationPage, error)
}

ProviderRegistry lists the storage providers a replacement can move to. It is an interface so provider selection stays testable without live observability.

type Server 

type Server struct {
	// contains filtered or unexported fields
}

Server provides /healthz and /metrics endpoints on a separate port.

func New 

func New(
	addr string,
	db *bun.DB,
	c cache.Cache,
	cacheGate *cacheaccess.Gate,
	cacheAccessTracker *cacheaccess.Tracker,
	cacheMaxBytes int64,
	repos *repository.Repositories,
	wh WorkerHealthChecker,
	wallet synapse.WalletQuerier,
	filecoinDefaultCopies int,
	logger *slog.Logger,
) *Server

New creates a new admin HTTP server.

func NewSetup 

func NewSetup(addr string, settings *SettingsService, logger *slog.Logger) *Server

NewSetup creates an admin HTTP server for first-run/setup mode.

func (*Server) Run 

func (s *Server) Run(ctx context.Context) error

Run starts the admin HTTP server at its configured address.

func (*Server) Serve 

func (s *Server) Serve(ctx context.Context, listener net.Listener) error

Serve serves the admin API on listener until ctx is cancelled. The listener is owned by Serve and closed during shutdown.

func (*Server) WithAuthConfig 

func (s *Server) WithAuthConfig(cfg config.AdminAuthConfig) error

func (*Server) WithEventHub 

func (s *Server) WithEventHub(events *EventHub) *Server

func (*Server) WithEvictMaxRetries 

func (s *Server) WithEvictMaxRetries(maxRetries int) *Server

WithEvictMaxRetries configures max retries for eviction tasks created by admin actions.

func (*Server) WithFilecoinReadiness 

func (s *Server) WithFilecoinReadiness(probe filecoinReadinessProbe) *Server

WithFilecoinReadiness enables Filecoin readiness API routes.

func (*Server) WithObjectStorage 

func (s *Server) WithObjectStorage(storage synapse.StorageClient) *Server

WithObjectStorage enables provider-backed object reads for admin downloads.

func (*Server) WithObjectUploader 

func (s *Server) WithObjectUploader(uploader objectUploader) *Server

WithObjectUploader enables admin object uploads through the existing object write path.

func (*Server) WithObjectVersionRestorer 

func (s *Server) WithObjectVersionRestorer(restorer objectVersionRestorer) *Server

WithObjectVersionRestorer enables admin object version restore operations.

func (*Server) WithObservability 

func (s *Server) WithObservability(service *observability.Service) *Server

func (*Server) WithProviderIdentityResolver 

func (s *Server) WithProviderIdentityResolver(resolver providerIdentityLookup) *Server

WithProviderIdentityResolver enables provider identity enrichment for admin APIs.

func (*Server) WithProviderReplacement 

func (s *Server) WithProviderReplacement(selector providerReplacementSelector) *Server

WithProviderReplacement enables the provider replacement endpoints. Without it they report that the storage service is unavailable.

func (*Server) WithProviderReplacementMaxRetries 

func (s *Server) WithProviderReplacementMaxRetries(maxRetries int) *Server

WithProviderReplacementMaxRetries sets the provider replacement retry limit.

func (*Server) WithS3IAM 

func (s *Server) WithS3IAM(iam auth.IAMService, rootAccess string) *Server

WithS3IAM enables S3 user management API routes.

func (*Server) WithSettings 

func (s *Server) WithSettings(settings *SettingsService) *Server

WithSettings enables settings API routes on a full admin server.

func (*Server) WithStorageCleanupMaxRetries 

func (s *Server) WithStorageCleanupMaxRetries(maxRetries int) *Server

WithStorageCleanupMaxRetries configures max retries for storage cleanup tasks created by admin actions.

func (*Server) WithTaskDiagnosticStatusChecker 

func (s *Server) WithTaskDiagnosticStatusChecker(checker taskDiagnosticStatusChecker) *Server

func (*Server) WithTrustedProxies 

func (s *Server) WithTrustedProxies(values []string) error

func (*Server) WithUploadMaxRetries 

func (s *Server) WithUploadMaxRetries(maxRetries int) *Server

WithUploadMaxRetries sets the retry budget replacement coordinators inherit.

type SettingsService 

type SettingsService struct {
	// contains filtered or unexported fields
}

func NewSettingsService 

func NewSettingsService(effective *config.Config, source config.Source) (*SettingsService, error)

func (*SettingsService) FilecoinDraftConfig 

func (s *SettingsService) FilecoinDraftConfig(req *settingsFilecoinUpdate) (*config.Config, []config.FieldError)

func (*SettingsService) Snapshot 

func (s *SettingsService) Snapshot(writable bool) settingsResponse

func (*SettingsService) Update 

func (s *SettingsService) Update(req settingsUpdateRequest, writable bool) (settingsResponse, []config.FieldError, error)

func (*SettingsService) Validate 

func (s *SettingsService) Validate(req settingsUpdateRequest) ([]config.FieldError, error)

type StorageProviderSelector 

type StorageProviderSelector struct {
	// contains filtered or unexported fields
}

StorageProviderSelector reports which storage providers a replacement can use. It resolves candidates only; the paid service itself is created later by the worker, after the operator has confirmed.

func NewStorageProviderSelector 

func NewStorageProviderSelector(providers ProviderRegistry) *StorageProviderSelector

NewStorageProviderSelector builds the provider inventory used by replacement.

func (*StorageProviderSelector) ListReplacementProviders 

func (s *StorageProviderSelector) ListReplacementProviders(ctx context.Context) ([]idtypes.OnChainID, error)

ListReplacementProviders returns the providers currently reachable and able to take a data set, lowest registry ID first so the same inventory always yields the same choice. Which of them a given replica can take is decided by replacementProviderCandidates, which owns that rule alone.

It reads the same provider observations the storage topology reports and applies the same availability filter, so the two views cannot disagree: a provider an operator sees listed as available there is offered here, and one that is unreachable is offered in neither. Reading the registry directly would drop the health signal and offer providers that cannot answer.

The inventory is deliberately not narrowed to the storage service's approved-provider list. That list governs the SDK's automatic selection only -- naming a provider outright bypasses it -- so filtering on it here would hide providers replacement can in fact use, and it is small enough for one bucket to exhaust.

type WorkerHealthChecker 

type WorkerHealthChecker interface {
	WorkerHealth() map[string]bool
}

WorkerHealthChecker provides worker liveness info. Implemented by worker.Manager.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.