diff --git a/pkg/config/datastore.go b/pkg/config/datastore.go index 5ec1c9b7..e90b95b4 100644 --- a/pkg/config/datastore.go +++ b/pkg/config/datastore.go @@ -32,8 +32,8 @@ const ( sbiNETCONF = "netconf" sbiGNMI = "gnmi" - ncCommitDatastoreRunning = "running" - ncCommitDatastoreCandidate = "candidate" + NcCommitDatastoreRunning = "running" + NcCommitDatastoreCandidate = "candidate" ) type DatastoreConfig struct { @@ -189,12 +189,12 @@ func (s *SBI) validateSetDefaults() error { case sbiNETCONF: switch s.NetconfOptions.CommitDatastore { case "": - s.NetconfOptions.CommitDatastore = ncCommitDatastoreCandidate - case ncCommitDatastoreRunning: - case ncCommitDatastoreCandidate: + s.NetconfOptions.CommitDatastore = NcCommitDatastoreCandidate + case NcCommitDatastoreRunning: + case NcCommitDatastoreCandidate: default: return fmt.Errorf("unknown commit-datastore: %s. Must be one of %s, %s", - s.NetconfOptions.CommitDatastore, ncCommitDatastoreCandidate, ncCommitDatastoreRunning) + s.NetconfOptions.CommitDatastore, NcCommitDatastoreCandidate, NcCommitDatastoreRunning) } case sbiGNMI: if s.GnmiOptions.Encoding == "" { diff --git a/pkg/core/core.go b/pkg/core/core.go new file mode 100644 index 00000000..5d496b9c --- /dev/null +++ b/pkg/core/core.go @@ -0,0 +1,190 @@ +// Package core is the in-process data-server: it owns the schema store, cache +// and datastores, and exposes them through the domain types of pkg/api. +package core + +import ( + "context" + "errors" + "fmt" + "sync" + + "github.com/sdcio/data-server/pkg/api" + "github.com/sdcio/data-server/pkg/cache" + "github.com/sdcio/data-server/pkg/config" + "github.com/sdcio/data-server/pkg/datastore" + "github.com/sdcio/data-server/pkg/schema" + logf "github.com/sdcio/logger" + "google.golang.org/grpc" +) + +// Core is the data-server engine. Create it with New, run Start once, wait for +// readiness with WaitReady and release it with Stop. It needs no gRPC listener +// or peer. +type Core struct { + cfg *config.Config + dialOpts []grpc.DialOption + + ctx context.Context + cancel context.CancelFunc + + startOnce sync.Once + stopOnce sync.Once + ready chan struct{} + readyErr error + + schemaClient schema.Client + cacheClient cache.Client + + mu sync.RWMutex + datastores map[string]*datastore.Datastore +} + +// Option customises a Core at construction. +type Option func(*Core) + +// WithDialOptions sets the gRPC dial options used for southbound gNMI targets. +func WithDialOptions(opts ...grpc.DialOption) Option { + return func(c *Core) { c.dialOpts = append(c.dialOpts, opts...) } +} + +// New builds a Core from a validated config. It starts no work; the Core's +// lifetime is bounded by ctx and Stop. +func New(ctx context.Context, cfg *config.Config, opts ...Option) (*Core, error) { + if cfg == nil { + return nil, errors.New("core: nil config") + } + ctx, cancel := context.WithCancel(ctx) + c := &Core{ + cfg: cfg, + ctx: ctx, + cancel: cancel, + ready: make(chan struct{}), + datastores: map[string]*datastore.Datastore{}, + } + for _, o := range opts { + o(c) + } + return c, nil +} + +// Start initialises the schema store, the cache and the configured datastores, +// then marks the Core ready. It returns when initialisation finished or failed +// and only the first call does work; a failure is also reported by WaitReady. +// Any configured datastore that cannot be created fails Start. +func (c *Core) Start(ctx context.Context) error { + c.startOnce.Do(func() { + c.readyErr = c.start(ctx) + close(c.ready) + }) + <-c.ready + return c.readyErr +} + +// WaitReady blocks until Start completed, returning its error, or until ctx is +// done. +func (c *Core) WaitReady(ctx context.Context) error { + select { + case <-c.ready: + return c.readyErr + case <-ctx.Done(): + return ctx.Err() + } +} + +// Stop stops all datastores and cancels the Core's context. It is idempotent +// and always returns nil: datastore stop errors are discarded. A Start after +// Stop fails with context.Canceled. +func (c *Core) Stop(ctx context.Context) error { + c.stopOnce.Do(func() { + c.mu.Lock() + defer c.mu.Unlock() + for _, ds := range c.datastores { + _ = ds.Stop(ctx) + } + c.cancel() + }) + return nil +} + +func (c *Core) start(ctx context.Context) error { + if err := c.ctx.Err(); err != nil { + return err + } + log := logf.FromContext(ctx).WithName("data-server") + ctx = logf.IntoContext(ctx, log) + + var wg sync.WaitGroup + var schemaErr, cacheErr error + wg.Add(2) + go func() { + defer wg.Done() + schemaErr = c.createSchemaClient(ctx) + }() + go func() { + defer wg.Done() + cacheErr = c.createCacheClient(ctx) + }() + wg.Wait() + if err := errors.Join(schemaErr, cacheErr); err != nil { + return fmt.Errorf("core init: %w", err) + } + + if err := c.createInitialDatastores(ctx); err != nil { + return fmt.Errorf("core init: %w", err) + } + log.Info("data-server ready") + return nil +} + +// createInitialDatastores creates the datastores listed in the config and +// returns the joined errors of those that failed. +func (c *Core) createInitialDatastores(ctx context.Context) error { + log := logf.FromContext(ctx) + var wg sync.WaitGroup + errs := make([]error, len(c.cfg.Datastores)) + for i, dsCfg := range c.cfg.Datastores { + dsCfg.Validation = c.cfg.Validation.DeepCopy() + wg.Add(1) + go func(i int, dsCfg *config.DatastoreConfig) { + defer wg.Done() + ds, err := datastore.New(c.ctx, dsCfg, c.schemaClient, c.cacheClient, c.dialOpts...) + if err == nil { + err = c.addDatastore(ds) + } + if err != nil { + errs[i] = fmt.Errorf("datastore %s: %w", dsCfg.Name, err) + } + }(i, dsCfg) + } + wg.Wait() + if err := errors.Join(errs...); err != nil { + return err + } + log.Info("configured datastores initialized") + return nil +} + +func (c *Core) addDatastore(ds *datastore.Datastore) error { + c.mu.Lock() + defer c.mu.Unlock() + if _, ok := c.datastores[ds.Name()]; ok { + return fmt.Errorf("datastore %s already exists", ds.Name()) + } + c.datastores[ds.Name()] = ds + return nil +} + +// getDatastore fails with api.ErrInvalidArgument for an empty name and +// api.ErrUnknownDatastore when the name is not registered. +func (c *Core) getDatastore(name string) (*datastore.Datastore, error) { + if name == "" { + return nil, fmt.Errorf("%w: missing datastore name", api.ErrInvalidArgument) + } + c.mu.RLock() + defer c.mu.RUnlock() + ds, ok := c.datastores[name] + if !ok { + return nil, fmt.Errorf("%w %s", api.ErrUnknownDatastore, name) + } + return ds, nil +} diff --git a/pkg/core/core_test.go b/pkg/core/core_test.go new file mode 100644 index 00000000..8c4e7e78 --- /dev/null +++ b/pkg/core/core_test.go @@ -0,0 +1,159 @@ +package core + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/sdcio/data-server/pkg/api" + "github.com/sdcio/data-server/pkg/config" + schemaConfig "github.com/sdcio/schema-server/pkg/config" +) + +func newTestConfig(t *testing.T) *config.Config { + t.Helper() + cfg, err := config.New("") + if err != nil { + t.Fatal(err) + } + cfg.SchemaStore = &schemaConfig.SchemaStoreConfig{ + Type: schemaConfig.StoreTypeMemory, + Schemas: []*schemaConfig.SchemaConfig{{ + Name: "s", Vendor: "v", Version: "1", + Files: []string{"../../tests/schema/sdcio_model_if.yang"}, + }}, + } + cfg.Cache = &config.CacheConfig{Type: "local", StoreType: "badgerdbsharded", Dir: t.TempDir()} + cfg.Datastores = []*config.DatastoreConfig{{ + Name: "ds1", + Schema: &config.SchemaConfig{Name: "s", Vendor: "v", Version: "1"}, + SBI: &config.SBI{Type: "noop"}, + Sync: &config.Sync{}, + }} + for _, ds := range cfg.Datastores { + if err := ds.ValidateSetDefaults(); err != nil { + t.Fatal(err) + } + } + return cfg +} + +func startCore(t *testing.T) *Core { + t.Helper() + ctx := context.Background() + c, err := New(ctx, newTestConfig(t)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = c.Stop(ctx) }) + go func() { _ = c.Start(ctx) }() + + wctx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + if err := c.WaitReady(wctx); err != nil { + t.Fatal(err) + } + return c +} + +func TestGetDataStoreInProcess(t *testing.T) { + c := startCore(t) + + info, err := c.GetDataStore(context.Background(), "ds1") + if err != nil { + t.Fatal(err) + } + if info.Name != "ds1" || info.Spec.Name != "ds1" { + t.Fatalf("unexpected info: %+v", info) + } + if info.Spec.Target.Type != api.TargetType("noop") { + t.Fatalf("target type = %q", info.Spec.Target.Type) + } + if info.Spec.Schema != (api.SchemaRef{Vendor: "v", Version: "1"}) { + t.Fatalf("schema = %+v", info.Spec.Schema) + } +} + +func TestGetDataStoreErrors(t *testing.T) { + c := startCore(t) + ctx := context.Background() + + if _, err := c.GetDataStore(ctx, ""); !errors.Is(err, api.ErrInvalidArgument) { + t.Fatalf("empty name: %v", err) + } + if _, err := c.GetDataStore(ctx, "nope"); !errors.Is(err, api.ErrUnknownDatastore) { + t.Fatalf("unknown: %v", err) + } +} + +func TestStopIsIdempotent(t *testing.T) { + c := startCore(t) + ctx := context.Background() + if err := c.Stop(ctx); err != nil { + t.Fatal(err) + } + if err := c.Stop(ctx); err != nil { + t.Fatal(err) + } +} + +func TestWatchDeviationsClosesOnCancel(t *testing.T) { + c := startCore(t) + ctx, cancel := context.WithCancel(context.Background()) + ch, err := c.WatchDeviations(ctx, "ds1") + if err != nil { + t.Fatal(err) + } + cancel() + select { + case _, ok := <-ch: + if ok { + t.Fatal("unexpected event") + } + case <-time.After(5 * time.Second): + t.Fatal("channel not closed") + } + if _, err := c.WatchDeviations(context.Background(), "nope"); !errors.Is(err, api.ErrUnknownDatastore) { + t.Fatalf("unknown: %v", err) + } +} + +func TestWaitReadyHonoursContext(t *testing.T) { + c, err := New(context.Background(), newTestConfig(t)) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if err := c.WaitReady(ctx); !errors.Is(err, context.Canceled) { + t.Fatalf("got %v", err) + } +} + +func TestStartAfterStopFails(t *testing.T) { + c, err := New(context.Background(), newTestConfig(t)) + if err != nil { + t.Fatal(err) + } + _ = c.Stop(context.Background()) + if err := c.Start(context.Background()); !errors.Is(err, context.Canceled) { + t.Fatalf("got %v", err) + } +} + +func TestStartFailsWhenInitialDatastoreFails(t *testing.T) { + cfg := newTestConfig(t) + cfg.Datastores[0].Schema.Name = "missing" + c, err := New(context.Background(), cfg) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = c.Stop(context.Background()) }) + if err := c.Start(context.Background()); err == nil { + t.Fatal("expected error") + } + if err := c.WaitReady(context.Background()); err == nil { + t.Fatal("WaitReady should report the failure") + } +} diff --git a/pkg/core/datastore.go b/pkg/core/datastore.go new file mode 100644 index 00000000..48c64667 --- /dev/null +++ b/pkg/core/datastore.go @@ -0,0 +1,150 @@ +package core + +import ( + "context" + "strings" + + "github.com/sdcio/data-server/pkg/api" + "github.com/sdcio/data-server/pkg/config" + "github.com/sdcio/data-server/pkg/datastore" + "github.com/sdcio/data-server/pkg/utils" + sdcpb "github.com/sdcio/sdc-protos/sdcpb" +) + +// GetDataStore returns the observed state of the named datastore, with +// credentials in the effective spec redacted. It fails with +// api.ErrInvalidArgument for an empty name and api.ErrUnknownDatastore when no +// such datastore exists. +func (c *Core) GetDataStore(ctx context.Context, name string) (*api.DatastoreInfo, error) { + ds, err := c.getDatastore(name) + if err != nil { + return nil, err + } + return datastoreInfo(ctx, ds) +} + +// WatchDeviations returns a channel of deviation events for the named +// datastore. The channel closes when ctx is done or the Core stops. It never +// emits events: no deviation producer feeds it. +func (c *Core) WatchDeviations(ctx context.Context, name string) (<-chan api.DeviationEvent, error) { + if _, err := c.getDatastore(name); err != nil { + return nil, err + } + ch := make(chan api.DeviationEvent, deviationBuffer) + go func() { + defer close(ch) + select { + case <-ctx.Done(): + case <-c.ctx.Done(): + } + }() + return ch, nil +} + +// deviationBuffer is the fixed capacity of a deviation watch channel. +const deviationBuffer = 8 + +func datastoreInfo(ctx context.Context, ds *datastore.Datastore) (*api.DatastoreInfo, error) { + intents, err := ds.IntentsList(ctx) + if err != nil { + return nil, err + } + state := ds.ConnectionState() + return &api.DatastoreInfo{ + Name: ds.Name(), + ConnectionState: connectionState(state.Status), + StatusDetails: state.Details, + Intents: intents, + Spec: specFromConfig(ds.Config()), + }, nil +} + +func connectionState(s sdcpb.TargetStatus) api.ConnectionState { + switch s { + case sdcpb.TargetStatus_CONNECTED: + return api.ConnectionConnected + case sdcpb.TargetStatus_NOT_CONNECTED: + return api.ConnectionNotConnected + default: + return api.ConnectionUnknown + } +} + +func specFromConfig(dc *config.DatastoreConfig) api.DatastoreSpec { + spec := api.DatastoreSpec{Name: dc.Name} + if dc.Schema != nil { + spec.Schema = api.SchemaRef{Vendor: dc.Schema.Vendor, Version: dc.Schema.Version} + } + if dc.SBI != nil { + spec.Target = targetFromSBI(dc.SBI) + } + if dc.Sync != nil { + for _, sp := range dc.Sync.Config { + spec.Sync.Configs = append(spec.Sync.Configs, api.SyncConfig{ + Name: sp.Name, + Target: &api.Target{Type: api.TargetType(strings.ToLower(sp.Protocol)), ProtocolOptions: api.ProtocolOptions{GNMI: gnmiSyncOptions(sp)}}, + Paths: sp.Paths, + Mode: syncMode(sp.Mode), + Interval: uint64(sp.Interval), + }) + } + } + return spec +} + +func gnmiSyncOptions(sp *config.SyncProtocol) *api.GNMIOptions { + if !strings.EqualFold(sp.Protocol, "gnmi") { + return nil + } + return &api.GNMIOptions{Encoding: api.GNMIEncoding(sp.Encoding)} +} + +func syncMode(m string) api.SyncMode { + switch m { + case "sample": + return api.SyncModeSample + case "once": + return api.SyncModeOnce + case "get": + return api.SyncModeGet + default: + return api.SyncModeOnChange + } +} + +func targetFromSBI(s *config.SBI) api.Target { + t := api.Target{ + Type: api.TargetType(strings.ToLower(s.Type)), + Address: s.Address, + Port: s.Port, + } + if s.Credentials != nil { + t.Credentials = &api.Credentials{Username: s.Credentials.Username, Password: redact(s.Credentials.Password), Token: redact(s.Credentials.Token)} + } + if s.TLS != nil { + t.TLS = &api.TLS{CA: s.TLS.CA, Cert: s.TLS.Cert, Key: s.TLS.Key, SkipVerify: s.TLS.SkipVerify} + } + if s.GnmiOptions != nil { + t.ProtocolOptions.GNMI = &api.GNMIOptions{Encoding: api.GNMIEncoding(s.GnmiOptions.Encoding), TargetName: s.GnmiOptions.TargetName} + } + if o := s.NetconfOptions; o != nil { + commit := api.CommitCandidateCandidate + if o.CommitDatastore == config.NcCommitDatastoreRunning { + commit = api.CommitCandidateRunning + } + t.ProtocolOptions.Netconf = &api.NetconfOptions{ + IncludeNs: o.IncludeNS, + OperationWithNs: o.OperationWithNamespace, + UseOperationRemove: o.UseOperationRemove, + CommitCandidate: commit, + } + } + return t +} + +func redact(v string) string { + if v == "" { + return "" + } + return utils.Redacted +} diff --git a/pkg/core/init.go b/pkg/core/init.go new file mode 100644 index 00000000..cff03cf2 --- /dev/null +++ b/pkg/core/init.go @@ -0,0 +1,122 @@ +package core + +import ( + "context" + "fmt" + "sync" + "time" + + cconfig "github.com/sdcio/cache/pkg/config" + "github.com/sdcio/data-server/pkg/cache" + "github.com/sdcio/data-server/pkg/schema" + logf "github.com/sdcio/logger" + schemaConfig "github.com/sdcio/schema-server/pkg/config" + schemaServerSchema "github.com/sdcio/schema-server/pkg/schema" + schemaStore "github.com/sdcio/schema-server/pkg/store" + schemaMemoryStore "github.com/sdcio/schema-server/pkg/store/memstore" + schemaPersistentStore "github.com/sdcio/schema-server/pkg/store/persiststore" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials" + "google.golang.org/grpc/credentials/insecure" +) + +func (c *Core) createCacheClient(ctx context.Context) error { + log := logf.FromContext(ctx).WithValues("cache-type", c.cfg.Cache.Type) + if c.cfg.Cache.Type != "local" { + return fmt.Errorf("unknown cache type: %s", c.cfg.Cache.Type) + } + log.Info("initializing local cache client", "store-type", c.cfg.Cache.StoreType, "directory", c.cfg.Cache.Dir) + cl, err := cache.NewLocalCache(&cconfig.CacheConfig{ + StoreType: c.cfg.Cache.StoreType, + Dir: c.cfg.Cache.Dir, + }) + if err != nil { + return fmt.Errorf("local cache: %w", err) + } + c.cacheClient = cl + log.Info("local cache created") + return nil +} + +func (c *Core) createSchemaClient(ctx context.Context) error { + if c.cfg.SchemaStore != nil { + return c.createLocalSchemaStore(ctx) + } + return c.createRemoteSchemaClient(ctx) +} + +// createLocalSchemaStore fails only when the store cannot be opened; a schema +// that fails to parse or add is logged and skipped. +func (c *Core) createLocalSchemaStore(ctx context.Context) error { + log := logf.FromContext(ctx) + var store schemaStore.Store + switch c.cfg.SchemaStore.Type { + case schemaConfig.StoreTypeMemory: + store = schemaMemoryStore.New() + case schemaConfig.StoreTypePersistent: + var err error + store, err = schemaPersistentStore.New(ctx, c.cfg.SchemaStore.Path, c.cfg.SchemaStore.Cache, c.cfg.SchemaStore.ReadOnly) + if err != nil { + return fmt.Errorf("persistent schema store: %w", err) + } + default: + return fmt.Errorf("unknown schema store type %q", c.cfg.SchemaStore.Type) + } + log.Info("parsing schema(s)...", "numschemas", len(c.cfg.SchemaStore.Schemas)) + + var wg sync.WaitGroup + for _, sCfg := range c.cfg.SchemaStore.Schemas { + wg.Add(1) + go func(sCfg *schemaConfig.SchemaConfig) { + defer wg.Done() + key := schemaStore.SchemaKey{Name: sCfg.Name, Vendor: sCfg.Vendor, Version: sCfg.Version} + log := log.WithValues("schema", key) + if store.HasSchema(key) { + log.Info("schema already exists in the store, not reloading it") + return + } + sc, err := schemaServerSchema.NewSchema(sCfg) + if err != nil { + log.Error(err, "schema parsing failed") + return + } + if err := store.AddSchema(sc); err != nil { + log.Error(err, "failed to add schema to the store") + } + }(sCfg) + } + wg.Wait() + c.schemaClient = schema.NewLocalClient(store) + return nil +} + +// createRemoteSchemaClient retries until the client is built or ctx is done. +func (c *Core) createRemoteSchemaClient(ctx context.Context) error { + log := logf.FromContext(ctx).WithValues("schema-server-address", c.cfg.SchemaServer.Address) + for { + cc, err := c.dialSchemaServer(ctx) + if err == nil { + log.Info("connected to schema server") + c.schemaClient = schema.NewRemoteClient(cc, c.cfg.SchemaServer.Cache) + return nil + } + log.Error(err, "failed to create schema client") + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(time.Second): + } + } +} + +func (c *Core) dialSchemaServer(ctx context.Context) (*grpc.ClientConn, error) { + creds := insecure.NewCredentials() + if c.cfg.SchemaServer.TLS != nil { + tlsCfg, err := c.cfg.SchemaServer.TLS.NewConfig(ctx) + if err != nil { + return nil, err + } + creds = credentials.NewTLS(tlsCfg) + } + return grpc.NewClient(c.cfg.SchemaServer.Address, grpc.WithTransportCredentials(creds)) +} diff --git a/pkg/datastore/target/netconf/nc.go b/pkg/datastore/target/netconf/nc.go index d46ba632..8b4774a5 100644 --- a/pkg/datastore/target/netconf/nc.go +++ b/pkg/datastore/target/netconf/nc.go @@ -165,7 +165,7 @@ func (t *ncTarget) Set(ctx context.Context, source types.TargetSource) (*sdcpb.S } switch t.sbiConfig.NetconfOptions.CommitDatastore { - case "running", "candidate": + case config.NcCommitDatastoreRunning, config.NcCommitDatastoreCandidate: return t.setToDevice(ctx, t.sbiConfig.NetconfOptions.CommitDatastore, source) } // should not get here if the config validation happened. @@ -274,7 +274,7 @@ func (t *ncTarget) setToDevice(ctx context.Context, commitDatastore string, sour } // candidate should discard on error - if commitDatastore == "candidate" { + if commitDatastore == config.NcCommitDatastoreCandidate { err2 := t.driver.Discard() if err2 != nil { // log failed discard @@ -289,7 +289,7 @@ func (t *ncTarget) setToDevice(ctx context.Context, commitDatastore string, sour } // candidate stores need to commit the changes to running - if commitDatastore == "candidate" { + if commitDatastore == config.NcCommitDatastoreCandidate { log.Info("committing changes on target") // commit the config err = t.driver.Commit() diff --git a/pkg/server/datastore.go b/pkg/server/datastore.go index 113b0590..f315ecf6 100644 --- a/pkg/server/datastore.go +++ b/pkg/server/datastore.go @@ -108,11 +108,11 @@ func (s *Server) CreateDataStore(ctx context.Context, req *sdcpb.CreateDataStore switch strings.ToLower(reqTarget.GetType()) { case "netconf": netconfOpts := reqTarget.GetNetconfOpts() - commitDatastore := "candidate" + commitDatastore := config.NcCommitDatastoreCandidate switch netconfOpts.GetCommitCandidate() { case sdcpb.CommitCandidate_COMMIT_CANDIDATE: case sdcpb.CommitCandidate_COMMIT_RUNNING: - commitDatastore = "running" + commitDatastore = config.NcCommitDatastoreRunning default: log.Error(nil, "unknown commitDatastore", "datastore", netconfOpts.GetCommitCandidate()) return nil, fmt.Errorf("unknown commitDatastore: %v", netconfOpts.GetCommitCandidate())