From 7ae766be8bcbc876d6dceb5fcf80ae9f7cdae00f Mon Sep 17 00:00:00 2001 From: Markus Vahlenkamp Date: Fri, 6 Nov 2026 14:59:50 +0100 Subject: [PATCH 1/4] feat(core): add in-process core with New/Start/Stop/WaitReady and GetDataStore Adds pkg/core, which builds the schema store, cache and datastores from a *config.Config without a gRPC listener or peer. GetDataStore returns api.DatastoreInfo and maps unknown/empty names to api sentinels. WatchDeviations fixes the channel API shape. Signed-off-by: Markus Vahlenkamp --- pkg/core/core.go | 169 ++++++++++++++++++++++++++++++++++++++++++ pkg/core/core_test.go | 132 +++++++++++++++++++++++++++++++++ pkg/core/datastore.go | 151 +++++++++++++++++++++++++++++++++++++ pkg/core/init.go | 120 ++++++++++++++++++++++++++++++ 4 files changed, 572 insertions(+) create mode 100644 pkg/core/core.go create mode 100644 pkg/core/core_test.go create mode 100644 pkg/core/datastore.go create mode 100644 pkg/core/init.go diff --git a/pkg/core/core.go b/pkg/core/core.go new file mode 100644 index 00000000..44a7d4d8 --- /dev/null +++ b/pkg/core/core.go @@ -0,0 +1,169 @@ +// 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/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 + cfn 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, + cfn: 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. +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. +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.cfn() + }) + return nil +} + +func (c *Core) start(ctx context.Context) error { + log := logf.FromContext(ctx) + + 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) + } + + c.createInitialDatastores(ctx) + log.Info("data-server ready") + return nil +} + +func (c *Core) createInitialDatastores(ctx context.Context) { + log := logf.FromContext(ctx) + var wg sync.WaitGroup + for _, dsCfg := range c.cfg.Datastores { + dsCfg.Validation = c.cfg.Validation.DeepCopy() + wg.Add(1) + go func(dsCfg *config.DatastoreConfig) { + defer wg.Done() + ds, err := datastore.New(c.ctx, dsCfg, c.schemaClient, c.cacheClient, c.dialOpts...) + if err != nil { + log.Error(err, "failed to create datastore", "datastore-name", dsCfg.Name) + return + } + if err := c.addDatastore(ds); err != nil { + log.Error(err, "failed to add datastore", "datastore-name", dsCfg.Name) + } + }(dsCfg) + } + wg.Wait() + log.Info("configured datastores initialized") +} + +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 +} + +func (c *Core) getDatastore(name string) (*datastore.Datastore, error) { + c.mu.RLock() + defer c.mu.RUnlock() + ds, ok := c.datastores[name] + if !ok { + return nil, fmt.Errorf("%w %s", 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..8152cd66 --- /dev/null +++ b/pkg/core/core_test.go @@ -0,0 +1,132 @@ +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) + } +} diff --git a/pkg/core/datastore.go b/pkg/core/datastore.go new file mode 100644 index 00000000..e799c9f6 --- /dev/null +++ b/pkg/core/datastore.go @@ -0,0 +1,151 @@ +package core + +import ( + "context" + "fmt" + "strings" + "time" + + "github.com/sdcio/data-server/pkg/api" + "github.com/sdcio/data-server/pkg/config" + "github.com/sdcio/data-server/pkg/datastore" + sdcpb "github.com/sdcio/sdc-protos/sdcpb" +) + +var errUnknownDatastore = api.ErrUnknownDatastore + +// GetDataStore returns the observed state of the named datastore. 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) { + if name == "" { + return nil, fmt.Errorf("%w: missing datastore name", api.ErrInvalidArgument) + } + 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. Events are +// not produced yet; the datastore deviation feed is wired in a later step. +func (c *Core) WatchDeviations(ctx context.Context, name string) (<-chan api.DeviationEvent, error) { + if name == "" { + return nil, fmt.Errorf("%w: missing datastore name", api.ErrInvalidArgument) + } + 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 / time.Nanosecond), + }) + } + } + 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: s.Credentials.Password, Token: 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 { + cc := api.CommitCandidateCandidate + if o.CommitDatastore == "running" { + cc = api.CommitCandidateRunning + } + t.ProtocolOptions.Netconf = &api.NetconfOptions{ + IncludeNs: o.IncludeNS, + OperationWithNs: o.OperationWithNamespace, + UseOperationRemove: o.UseOperationRemove, + CommitCandidate: cc, + } + } + return t +} diff --git a/pkg/core/init.go b/pkg/core/init.go new file mode 100644 index 00000000..a97a2240 --- /dev/null +++ b/pkg/core/init.go @@ -0,0 +1,120 @@ +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) +} + +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() + sck := schemaStore.SchemaKey{Name: sCfg.Name, Vendor: sCfg.Vendor, Version: sCfg.Version} + log := log.WithValues("schema", sck) + if store.HasSchema(sck) { + 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)) +} From 2219bdfcd833742d6b9fec929c545958cd4e03a3 Mon Sep 17 00:00:00 2001 From: Markus Vahlenkamp Date: Fri, 6 Nov 2026 15:01:57 +0100 Subject: [PATCH 2/4] refactor(core): redact credentials, fail Start after Stop, tidy docs and errors Names the core logger data-server, validates the datastore name in one place, drops the error alias and documents swallowed errors. Signed-off-by: Markus Vahlenkamp --- pkg/core/core.go | 26 +++++++++++++++++++------- pkg/core/core_test.go | 11 +++++++++++ pkg/core/datastore.go | 39 +++++++++++++++++++-------------------- pkg/core/init.go | 8 +++++--- 4 files changed, 54 insertions(+), 30 deletions(-) diff --git a/pkg/core/core.go b/pkg/core/core.go index 44a7d4d8..1d17a173 100644 --- a/pkg/core/core.go +++ b/pkg/core/core.go @@ -8,6 +8,7 @@ import ( "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" @@ -23,8 +24,8 @@ type Core struct { cfg *config.Config dialOpts []grpc.DialOption - ctx context.Context - cfn context.CancelFunc + ctx context.Context + cancel context.CancelFunc startOnce sync.Once stopOnce sync.Once @@ -56,7 +57,7 @@ func New(ctx context.Context, cfg *config.Config, opts ...Option) (*Core, error) c := &Core{ cfg: cfg, ctx: ctx, - cfn: cancel, + cancel: cancel, ready: make(chan struct{}), datastores: map[string]*datastore.Datastore{}, } @@ -89,7 +90,9 @@ func (c *Core) WaitReady(ctx context.Context) error { } } -// Stop stops all datastores and cancels the Core's context. It is idempotent. +// 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() @@ -97,13 +100,17 @@ func (c *Core) Stop(ctx context.Context) error { for _, ds := range c.datastores { _ = ds.Stop(ctx) } - c.cfn() + c.cancel() }) return nil } func (c *Core) start(ctx context.Context) error { - log := logf.FromContext(ctx) + 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 @@ -158,12 +165,17 @@ func (c *Core) addDatastore(ds *datastore.Datastore) error { 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", errUnknownDatastore, name) + 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 index 8152cd66..ff7f094c 100644 --- a/pkg/core/core_test.go +++ b/pkg/core/core_test.go @@ -130,3 +130,14 @@ func TestWaitReadyHonoursContext(t *testing.T) { 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) + } +} diff --git a/pkg/core/datastore.go b/pkg/core/datastore.go index e799c9f6..8a4e9268 100644 --- a/pkg/core/datastore.go +++ b/pkg/core/datastore.go @@ -2,25 +2,20 @@ package core import ( "context" - "fmt" "strings" - "time" "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" ) -var errUnknownDatastore = api.ErrUnknownDatastore - -// GetDataStore returns the observed state of the named datastore. It fails -// with api.ErrInvalidArgument for an empty name and api.ErrUnknownDatastore -// when no such datastore exists. +// 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) { - if name == "" { - return nil, fmt.Errorf("%w: missing datastore name", api.ErrInvalidArgument) - } ds, err := c.getDatastore(name) if err != nil { return nil, err @@ -29,12 +24,9 @@ func (c *Core) GetDataStore(ctx context.Context, name string) (*api.DatastoreInf } // WatchDeviations returns a channel of deviation events for the named -// datastore. The channel closes when ctx is done or the Core stops. Events are -// not produced yet; the datastore deviation feed is wired in a later step. +// 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 name == "" { - return nil, fmt.Errorf("%w: missing datastore name", api.ErrInvalidArgument) - } if _, err := c.getDatastore(name); err != nil { return nil, err } @@ -93,7 +85,7 @@ func specFromConfig(dc *config.DatastoreConfig) api.DatastoreSpec { 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 / time.Nanosecond), + Interval: uint64(sp.Interval), }) } } @@ -127,7 +119,7 @@ func targetFromSBI(s *config.SBI) api.Target { Port: s.Port, } if s.Credentials != nil { - t.Credentials = &api.Credentials{Username: s.Credentials.Username, Password: s.Credentials.Password, Token: s.Credentials.Token} + 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} @@ -136,16 +128,23 @@ func targetFromSBI(s *config.SBI) api.Target { t.ProtocolOptions.GNMI = &api.GNMIOptions{Encoding: api.GNMIEncoding(s.GnmiOptions.Encoding), TargetName: s.GnmiOptions.TargetName} } if o := s.NetconfOptions; o != nil { - cc := api.CommitCandidateCandidate + commit := api.CommitCandidateCandidate if o.CommitDatastore == "running" { - cc = api.CommitCandidateRunning + commit = api.CommitCandidateRunning } t.ProtocolOptions.Netconf = &api.NetconfOptions{ IncludeNs: o.IncludeNS, OperationWithNs: o.OperationWithNamespace, UseOperationRemove: o.UseOperationRemove, - CommitCandidate: cc, + 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 index a97a2240..cff03cf2 100644 --- a/pkg/core/init.go +++ b/pkg/core/init.go @@ -45,6 +45,8 @@ func (c *Core) createSchemaClient(ctx context.Context) error { 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 @@ -67,9 +69,9 @@ func (c *Core) createLocalSchemaStore(ctx context.Context) error { wg.Add(1) go func(sCfg *schemaConfig.SchemaConfig) { defer wg.Done() - sck := schemaStore.SchemaKey{Name: sCfg.Name, Vendor: sCfg.Vendor, Version: sCfg.Version} - log := log.WithValues("schema", sck) - if store.HasSchema(sck) { + 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 } From 38c20367718987147d5c34df66b1d79c85d732fa Mon Sep 17 00:00:00 2001 From: Markus Vahlenkamp Date: Fri, 6 Nov 2026 15:06:01 +0100 Subject: [PATCH 3/4] feat(core): fail Start when a configured datastore cannot be created Signed-off-by: Markus Vahlenkamp --- pkg/core/core.go | 29 +++++++++++++++++++---------- pkg/core/core_test.go | 16 ++++++++++++++++ 2 files changed, 35 insertions(+), 10 deletions(-) diff --git a/pkg/core/core.go b/pkg/core/core.go index 1d17a173..5d496b9c 100644 --- a/pkg/core/core.go +++ b/pkg/core/core.go @@ -70,6 +70,7 @@ func New(ctx context.Context, cfg *config.Config, opts ...Option) (*Core, error) // 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) @@ -128,31 +129,39 @@ func (c *Core) start(ctx context.Context) error { return fmt.Errorf("core init: %w", err) } - c.createInitialDatastores(ctx) + if err := c.createInitialDatastores(ctx); err != nil { + return fmt.Errorf("core init: %w", err) + } log.Info("data-server ready") return nil } -func (c *Core) createInitialDatastores(ctx context.Context) { +// 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 - for _, dsCfg := range c.cfg.Datastores { + 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(dsCfg *config.DatastoreConfig) { + 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 { - log.Error(err, "failed to create datastore", "datastore-name", dsCfg.Name) - return + if err == nil { + err = c.addDatastore(ds) } - if err := c.addDatastore(ds); err != nil { - log.Error(err, "failed to add datastore", "datastore-name", dsCfg.Name) + if err != nil { + errs[i] = fmt.Errorf("datastore %s: %w", dsCfg.Name, err) } - }(dsCfg) + }(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 { diff --git a/pkg/core/core_test.go b/pkg/core/core_test.go index ff7f094c..8c4e7e78 100644 --- a/pkg/core/core_test.go +++ b/pkg/core/core_test.go @@ -141,3 +141,19 @@ func TestStartAfterStopFails(t *testing.T) { 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") + } +} From acaa06d11c56fddec3256de60816135db3c40269 Mon Sep 17 00:00:00 2001 From: Markus Vahlenkamp Date: Wed, 11 Nov 2026 16:45:26 +0100 Subject: [PATCH 4/4] Use exported constants for netconf commit-datastore values Signed-off-by: Markus Vahlenkamp --- pkg/config/datastore.go | 12 ++++++------ pkg/core/datastore.go | 2 +- pkg/datastore/target/netconf/nc.go | 6 +++--- pkg/server/datastore.go | 4 ++-- 4 files changed, 12 insertions(+), 12 deletions(-) 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/datastore.go b/pkg/core/datastore.go index 8a4e9268..48c64667 100644 --- a/pkg/core/datastore.go +++ b/pkg/core/datastore.go @@ -129,7 +129,7 @@ func targetFromSBI(s *config.SBI) api.Target { } if o := s.NetconfOptions; o != nil { commit := api.CommitCandidateCandidate - if o.CommitDatastore == "running" { + if o.CommitDatastore == config.NcCommitDatastoreRunning { commit = api.CommitCandidateRunning } t.ProtocolOptions.Netconf = &api.NetconfOptions{ 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())