server

package
v0.0.0-...-165bdb7 Latest Latest
Warning

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

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

Documentation

Overview

Package server implements the MonoFS gRPC server with NutsDB storage.

Package server implements the MonoFS gRPC server with NutsDB storage.

Package server implements the MonoFS gRPC server.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type BlobMeta 

type BlobMeta struct {
	BlobHash   string
	RepoURL    string
	Branch     string
	SourceType fetcher.SourceType
	ModulePath string // For Go modules
	Version    string // For Go modules
}

BlobMeta contains metadata for prefetch requests.

type CfgBackendStore 

type CfgBackendStore interface {
	// kvsapi.ReadStore
	ReadFile(ctx context.Context, logicalPath string) ([]byte, error)
	ListDir(ctx context.Context, logicalDir string) ([]kvsapi.DirEntry, error)
	Stat(ctx context.Context, logicalPath string) (kvsapi.FileInfo, error)

	// kvsapi.WatchStore
	Watch(ctx context.Context, prefixes []string) (<-chan kvsapi.ChangeEvent, error)

	// Config management
	PushVersion(ctx context.Context, product, sha string, files map[string][]byte) error
	SetCurrent(ctx context.Context, product, sha string) error
	GetCurrent(ctx context.Context, product string) (string, error)
	ListProductVersions(ctx context.Context, product string) ([]cfg.VersionRecord, error)
	Compact(ctx context.Context, product string) error
}

CfgBackendStore is the interface that CfgStore must satisfy for use in the MonoFS server. It exposes the read and watch interfaces of kvsapi.Store plus the cfg-specific management methods the server needs for ingestion.

type DoctorBackend 

type DoctorBackend interface {
	IngestLogs(ctx context.Context, chunkID string, logs []logengine.LogRecord) error
	IngestMetrics(ctx context.Context, chunkID string, metrics []logengine.MetricRecord) error
	IngestTraces(ctx context.Context, chunkID string, spans []logengine.SpanRecord) error
	StreamLogs(ctx context.Context, query, service string, from, to time.Time, limit int, yield func(logengine.LogRecord) error) error
	StreamMetrics(ctx context.Context, query logengine.MetricQuery, from, to time.Time, yield func(logengine.MetricRecord) error) error
	StreamTraces(ctx context.Context, traceID, service string, from, to time.Time, limit int, yield func(logengine.SpanRecord) error) error
	QueryLogs(ctx context.Context, query, service string, from, to time.Time, limit int) ([]logengine.LogRecord, error)
	QueryMetrics(ctx context.Context, query logengine.MetricQuery, from, to time.Time) ([]logengine.MetricRecord, error)
	QueryTraces(ctx context.Context, traceID, service string, from, to time.Time, limit int) ([]logengine.SpanRecord, error)
	Stats(ctx context.Context) (logengine.LogEngineStats, error)
}

DoctorBackend is the interface the server uses to ingest and query telemetry signals. *logengine.LogEngine satisfies this interface.

type KVSStore 

type KVSStore interface {
	kvsapi.Store
	Status() kvsapi.StoreStatus
	Close() error
}

type PredictedFile 

type PredictedFile struct {
	StorageID   string
	FilePath    string
	Probability float64
	Source      string // "markov", "directory", "structural"
	ContentID   string // Blob hash (filled by caller if known)
}

PredictedFile represents a prefetch candidate.

type Predictor 

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

Predictor implements predictive prefetching using Markov chains and clustering. It runs on storage nodes and tracks access patterns to predict future accesses.

func NewPredictor 

func NewPredictor(fetcherClient *fetcher.Client, config PredictorConfig, logger *slog.Logger) *Predictor

NewPredictor creates a new prediction engine.

func (*Predictor) GetStats 

func (p *Predictor) GetStats() PredictorStats

GetStats returns predictor statistics.

func (*Predictor) Predict 

func (p *Predictor) Predict(storageID, filePath string) []PredictedFile

Predict returns predicted files based on access patterns.

func (*Predictor) RecordAccess 

func (p *Predictor) RecordAccess(ctx context.Context, storageID, filePath, clientID string, meta *BlobMeta)

RecordAccess records a file access and triggers prediction.

type PredictorConfig 

type PredictorConfig struct {
	// Markov chain settings
	MaxTransitionsPerFile int     // Max edges per source file
	TransitionDecayRate   float64 // Decay per hour (e.g., 0.95)
	MinTransitionCount    int     // Min transitions before predicting
	MarkovDepth           int     // How many hops to look ahead

	// Directory prediction
	DirectoryPrefetchSize int     // Max files to prefetch per directory
	DirectoryThreshold    float64 // Min access ratio to prefetch

	// Temporal settings
	SessionTimeout time.Duration // Gap that ends a session
	RecentWindow   int           // Number of recent accesses to track

	// Prefetch settings
	PrefetchThreshold float64 // Min probability to prefetch
	MaxPrefetchFiles  int     // Max files per prediction
	PrefetchPriority  int     // Priority for prefetch requests (0-10)

	// Cleanup
	CleanupInterval time.Duration
	MaxChainAge     time.Duration

	// Filtering
	IgnoreClientIDs []string // Client IDs to ignore (e.g., "search-indexer")
}

PredictorConfig configures the prediction engine.

func DefaultPredictorConfig 

func DefaultPredictorConfig() PredictorConfig

DefaultPredictorConfig returns sensible defaults.

type PredictorStats 

type PredictorStats struct {
	MarkovChains  int
	DirectoryMaps int
	Predictions   int64
	Prefetches    int64
	PrefetchHits  int64
}

PredictorStats holds prediction statistics.

type Server 

type Server struct {
	pb.UnimplementedMonoFSServer
	// contains filtered or unexported fields
}

Server implements the MonoFS gRPC server with NutsDB storage.

func NewServer 

func NewServer(nodeID, address, dbPath, gitCacheDir string, dbSync bool, logger *slog.Logger) (*Server, error)

NewServer creates a new MonoFS server.

func (*Server) AppendLedgerEntries 

func (s *Server) AppendLedgerEntries(_ context.Context, req *pb.AppendLedgerEntriesRequest) (*pb.AppendLedgerEntriesResponse, error)

AppendLedgerEntries appends commit/outcome/refresh records to the node-owned ledger.

func (*Server) Authenticate 

func (s *Server) Authenticate(ctx context.Context, req *pb.AuthRequest) (*pb.AuthResponse, error)

Authenticate implements the Authenticate RPC.

func (*Server) BuildDirectoryIndexes 

func (s *Server) BuildDirectoryIndexes(ctx context.Context, req *pb.BuildDirectoryIndexesRequest) (*pb.BuildDirectoryIndexesResponse, error)

BuildDirectoryIndexes builds directory indexes for all files in a repository. This is called after ingestion completes to improve ingestion performance.

func (*Server) ClearFailoverCache 

func (s *Server) ClearFailoverCache(ctx context.Context, req *pb.ClearFailoverCacheRequest) (*pb.ClearFailoverCacheResponse, error)

ClearFailoverCache removes temporary failover metadata after node recovery. Called by router when a failed node comes back online.

func (*Server) Close 

func (s *Server) Close() error

Close closes the server resources.

func (*Server) ConfigureFetcher 

func (s *Server) ConfigureFetcher(fetcherAddrs []string) error

ConfigureFetcher sets up the fetcher client for external blob retrieval. This MUST be called after NewServer - storage nodes require fetchers for blob access.

func (*Server) Create 

func (s *Server) Create(ctx context.Context, req *pb.CreateRequest) (*pb.CreateResponse, error)

Create implements the Create RPC.

func (*Server) DeleteDirectoryRecursive 

func (s *Server) DeleteDirectoryRecursive(ctx context.Context, req *pb.DeleteDirectoryRecursiveRequest) (*pb.DeleteDirectoryRecursiveResponse, error)

DeleteDirectoryRecursive removes a directory and all its contents from the node.

func (*Server) DeleteFile 

func (s *Server) DeleteFile(ctx context.Context, req *pb.DeleteFileRequest) (*pb.DeleteFileResponse, error)

DeleteFile removes a file's metadata after rebalancing (called by router). This is used to clean up old file copies after files have been moved to new nodes. Important: This is ONLY called during rebalancing cleanup, NOT during recovery.

func (*Server) DeleteRepository 

func (s *Server) DeleteRepository(ctx context.Context, req *pb.DeleteRepositoryOnNodeRequest) (*pb.DeleteRepositoryOnNodeResponse, error)

DeleteRepository removes all data for a repository from this node. This includes: repo info, display path lookup, onboarding status, all owned files, all replica files, all directory indexes.

func (*Server) DisableForwarding 

func (s *Server) DisableForwarding()

DisableForwarding stops the forwarding functionality and closes connections.

func (*Server) EnableForwarding 

func (s *Server) EnableForwarding(routerAddr string, refreshInterval time.Duration) error

EnableForwarding enables server-side request forwarding via HRW routing. The server will connect to the router, fetch cluster topology, and forward requests to the correct owner node when needed.

This makes any server act as a "smart proxy" that can handle any request by routing it to the appropriate backend node.

func (*Server) GetAttr 

func (s *Server) GetAttr(ctx context.Context, req *pb.GetAttrRequest) (*pb.GetAttrResponse, error)

GetAttr implements the GetAttr RPC.

func (*Server) GetForwardingStats 

func (s *Server) GetForwardingStats() map[string]interface{}

GetForwardingStats returns statistics about the forwarding functionality.

func (*Server) GetNodeInfo 

func (s *Server) GetNodeInfo(ctx context.Context, req *pb.NodeInfoRequest) (*pb.NodeInfoResponse, error)

GetNodeInfo implements the GetNodeInfo RPC. This is called frequently (every 10s) by the router for health checks, so we use a cached atomic counter instead of scanning the database.

func (*Server) GetOnboardingStatus 

func (s *Server) GetOnboardingStatus(ctx context.Context, req *pb.OnboardingStatusRequest) (*pb.OnboardingStatusResponse, error)

GetOnboardingStatus returns onboarding status for all repositories on this node.

func (*Server) GetPredictorStats 

func (s *Server) GetPredictorStats(ctx context.Context, req *pb.PredictorStatsRequest) (*pb.PredictorStatsResponse, error)

GetPredictorStats returns predictor statistics for this node via gRPC.

func (*Server) GetPrefetchStats 

func (s *Server) GetPrefetchStats() (hits, misses uint64)

GetPrefetchStats returns prefetch hit/miss statistics.

func (*Server) GetRepositoryFiles 

func (s *Server) GetRepositoryFiles(ctx context.Context, req *pb.GetRepositoryFilesRequest) (*pb.GetRepositoryFilesResponse, error)

GetRepositoryFiles returns list of files this node owns for a repository.

func (*Server) GetRepositoryInfo 

func (s *Server) GetRepositoryInfo(ctx context.Context, req *pb.GetRepositoryInfoRequest) (*pb.GetRepositoryInfoResponse, error)

GetRepositoryInfo returns metadata for a specific repository.

func (*Server) IngestFile 

func (s *Server) IngestFile(ctx context.Context, req *pb.IngestFileRequest) (*pb.IngestFileResponse, error)

IngestFile stores file metadata from router ingestion.

func (*Server) IngestFileBatch 

func (s *Server) IngestFileBatch(ctx context.Context, req *pb.IngestFileBatchRequest) (*pb.IngestFileBatchResponse, error)

IngestFileBatch stores multiple file metadata in a single database transaction. This is significantly faster than calling IngestFile repeatedly (10-50x improvement).

func (*Server) IngestLogs 

func (s *Server) IngestLogs(ctx context.Context, req *pb.IngestLogsRequest) (*pb.IngestLogsResponse, error)

IngestLogs implements the Doctor partition log ingest.

func (*Server) IngestMetrics 

func (s *Server) IngestMetrics(ctx context.Context, req *pb.IngestMetricsRequest) (*pb.IngestMetricsResponse, error)

IngestMetrics implements the Doctor partition metric ingest.

func (*Server) IngestReplicaBatch 

func (s *Server) IngestReplicaBatch(ctx context.Context, req *pb.IngestReplicaBatchRequest) (*pb.IngestReplicaBatchResponse, error)

IngestReplicaBatch stores replica metadata for failover purposes. Unlike IngestFileBatch (which stores primary ownership), this stores backup copies in bucketReplicaFiles so they can be used for instant failover.

func (*Server) IngestTraces 

func (s *Server) IngestTraces(ctx context.Context, req *pb.IngestTracesRequest) (*pb.IngestTracesResponse, error)

IngestTraces implements the Doctor partition trace ingest.

func (*Server) ListRepositories 

func (s *Server) ListRepositories(ctx context.Context, req *pb.ListRepositoriesRequest) (*pb.ListRepositoriesResponse, error)

ListRepositories returns all repository IDs stored on this node.

func (*Server) Lookup 

func (s *Server) Lookup(ctx context.Context, req *pb.LookupRequest) (*pb.LookupResponse, error)

Lookup implements the Lookup RPC.

func (*Server) MarkRepositoryOnboarded 

func (s *Server) MarkRepositoryOnboarded(ctx context.Context, req *pb.MarkRepositoryOnboardedRequest) (*pb.MarkRepositoryOnboardedResponse, error)

MarkRepositoryOnboarded marks a repository as fully onboarded.

func (*Server) NodeID 

func (s *Server) NodeID() string

NodeID returns the server's node ID.

func (*Server) QueryLedger 

func (s *Server) QueryLedger(_ context.Context, req *pb.QueryLedgerRequest) (*pb.QueryLedgerResponse, error)

QueryLedger returns filtered ledger records from the node-owned ledger.

func (*Server) QueryLogs 

func (s *Server) QueryLogs(ctx context.Context, req *pb.QueryLogsRequest) (*pb.QueryLogsResponse, error)

func (*Server) QueryMetrics 

func (s *Server) QueryMetrics(ctx context.Context, req *pb.QueryMetricsRequest) (*pb.QueryMetricsResponse, error)

func (*Server) QueryTraces 

func (s *Server) QueryTraces(ctx context.Context, req *pb.QueryTracesRequest) (*pb.QueryTracesResponse, error)

func (*Server) Read 

func (s *Server) Read(req *pb.ReadRequest, stream grpc.ServerStreamingServer[pb.DataChunk]) error

Read implements the Read RPC - lazy loads from Git repo.

func (*Server) ReadDir 

func (s *Server) ReadDir(req *pb.ReadDirRequest, stream grpc.ServerStreamingServer[pb.DirEntry]) error

ReadDir implements the ReadDir RPC (streaming).

func (*Server) Register 

func (s *Server) Register(grpcServer *grpc.Server)

Register registers the server with a gRPC server.

func (*Server) RegisterRepository 

func (s *Server) RegisterRepository(ctx context.Context, req *pb.RegisterRepositoryRequest) (*pb.RegisterRepositoryResponse, error)

RegisterRepository registers repository metadata on this node. This is called by the router BEFORE file ingestion to ensure all nodes know about the repository and can resolve display paths.

func (*Server) SetCfgStore 

func (s *Server) SetCfgStore(store CfgBackendStore)

SetCfgStore wires a CfgBackendStore into the server. Call this before serving requests if any repository uses the "cfg" backend.

func (*Server) SetDoctorBackend 

func (s *Server) SetDoctorBackend(b DoctorBackend)

SetDoctorBackend configures the telemetry backend for this server node.

func (*Server) SetKVSStore 

func (s *Server) SetKVSStore(store KVSStore)

func (*Server) StreamQueryLogs 

func (s *Server) StreamQueryLogs(req *pb.QueryLogsRequest, stream grpc.ServerStreamingServer[pb.QueryResultItem]) error

QueryLogs implements the Doctor partition log query.

func (*Server) StreamQueryMetrics 

func (s *Server) StreamQueryMetrics(req *pb.QueryMetricsRequest, stream grpc.ServerStreamingServer[pb.QueryResultItem]) error

QueryMetrics implements the Doctor partition metric query.

func (*Server) StreamQueryTraces 

func (s *Server) StreamQueryTraces(req *pb.QueryTracesRequest, stream grpc.ServerStreamingServer[pb.QueryResultItem]) error

QueryTraces implements the Doctor partition trace query.

func (*Server) StreamRepositoryFiles 

func (s *Server) StreamRepositoryFiles(req *pb.GetRepositoryFilesRequest, stream grpc.ServerStreamingServer[pb.RepositoryFileItem]) error

func (*Server) SyncMetadataFromNode 

func (s *Server) SyncMetadataFromNode(ctx context.Context, req *pb.SyncMetadataFromNodeRequest) (*pb.SyncMetadataFromNodeResponse, error)

SyncMetadataFromNode implements failover metadata synchronization. Called by router when this node becomes a backup for a failed node. Copies replica metadata into failover cache for fast lookup.

func (*Server) Write 

func (s *Server) Write(stream grpc.ClientStreamingServer[pb.WriteRequest, pb.WriteResponse]) error

Write implements the Write RPC (client streaming).

type StubServer 

type StubServer struct {
	pb.UnimplementedMonoFSServer
	// contains filtered or unexported fields
}

StubServer implements the MonoFS gRPC server with in-memory stub data.

func NewStubServer 

func NewStubServer(nodeID, address string, logger *slog.Logger) *StubServer

NewStubServer creates a new stub server with sample data.

func (*StubServer) Authenticate 

func (s *StubServer) Authenticate(ctx context.Context, req *pb.AuthRequest) (*pb.AuthResponse, error)

Authenticate implements the Authenticate RPC.

func (*StubServer) Create 

func (s *StubServer) Create(ctx context.Context, req *pb.CreateRequest) (*pb.CreateResponse, error)

Create implements the Create RPC.

func (*StubServer) GetAttr 

func (s *StubServer) GetAttr(ctx context.Context, req *pb.GetAttrRequest) (*pb.GetAttrResponse, error)

GetAttr implements the GetAttr RPC.

func (*StubServer) GetNodeInfo 

func (s *StubServer) GetNodeInfo(ctx context.Context, req *pb.NodeInfoRequest) (*pb.NodeInfoResponse, error)

GetNodeInfo implements the GetNodeInfo RPC.

func (*StubServer) Lookup 

func (s *StubServer) Lookup(ctx context.Context, req *pb.LookupRequest) (*pb.LookupResponse, error)

Lookup implements the Lookup RPC.

func (*StubServer) NodeID 

func (s *StubServer) NodeID() string

NodeID returns the server's node ID.

func (*StubServer) Read 

func (s *StubServer) Read(req *pb.ReadRequest, stream grpc.ServerStreamingServer[pb.DataChunk]) error

Read implements the Read RPC (streaming).

func (*StubServer) ReadDir 

func (s *StubServer) ReadDir(req *pb.ReadDirRequest, stream grpc.ServerStreamingServer[pb.DirEntry]) error

ReadDir implements the ReadDir RPC (streaming).

func (*StubServer) Register 

func (s *StubServer) Register(grpcServer *grpc.Server)

Register registers the server with a gRPC server.

func (*StubServer) Write 

func (s *StubServer) Write(stream grpc.ClientStreamingServer[pb.WriteRequest, pb.WriteResponse]) error

Write implements the Write RPC (client streaming).

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.