feat(core): serve the remaining RPCs in-process

pkg/api/transaction.go

@@ -1,6 +1,10 @@
 package api
 
-import "time"
+import (
+	"time"
+
+	sdcpb "github.com/sdcio/sdc-protos/sdcpb"
+)
 
 // Origin identifies the requester of an operation for logging and audit. It is
 // optional; the gRPC adapter fills it from the peer and in-process callers set
@@ -20,6 +24,9 @@ type SetOptions struct {
 	// Timeout auto-cancels an unconfirmed transaction; zero means the default.
 	Timeout time.Duration
 	Origin  Origin
+	// Replace, when set, replaces the whole device configuration with its
+	// content. Its name and priority are ignored.
+	Replace *Intent
 }
 
 // IntentSpec is the body of an Intent: exactly one of Update and Delete is set.
@@ -34,11 +41,22 @@ type Intent struct {
 	Spec IntentSpec
 }
 
+// UpdateEntry is the JSON content of an intent below one path.
+type UpdateEntry struct {
+	// Path addresses the content; nil or empty addresses the root.
+	Path *sdcpb.Path
+	// Content is the JSON encoded value at Path.
+	Content []byte
+}
+
 // Update creates or replaces an intent.
 type Update struct {
 	Priority int32
-	// Content is the intent body, JSON encoded.
-	Content []byte
+	Entries  []UpdateEntry
+	// Deletes are paths removed explicitly on behalf of the intent.
+	Deletes []*sdcpb.Path
+	// RevertPaths limits reverting to these paths for a non-revertive intent.
+	RevertPaths []*sdcpb.Path
 	// NonRevertive leaves device-side changes in place instead of reverting them.
 	NonRevertive bool
 	// PreviouslyApplied marks a non-revertive intent that was applied before.
@@ -71,3 +89,49 @@ type IntentRequest struct {
 	Format    Format
 	Origin    Origin
 }
+
+// IntentResult is an intent read back from the intended store. Content holds
+// the intent encoded as requested: JSON, JSON IETF, XML or protojson encoded
+// xpath/value pairs; for FormatProto it is a proto encoded sdcpb.Intent.
+type IntentResult struct {
+	Name         string
+	Priority     int32
+	Orphan       bool
+	NonRevertive bool
+	Deletes      []*sdcpb.Path
+	Format       Format
+	Content      []byte
+}
+
+// IntentValidation holds the findings of one intent in a transaction.
+type IntentValidation struct {
+	Warnings []string
+	Errors   []string
+}
+
+// PathValue is a path with its typed value.
+type PathValue struct {
+	Path  *sdcpb.Path
+	Value *sdcpb.TypedValue
+}
+
+// TransactionResult is the outcome of a transaction set. Validation failures
+// are reported in Intents, not as an error.
+type TransactionResult struct {
+	Updates  []PathValue
+	Deletes  []*sdcpb.Path
+	Intents  map[string]IntentValidation
+	Warnings []string
+}
+
+// BlameNode is one level of the blame tree: which intent owns the value below
+// a path. Values keep their original type.
+type BlameNode struct {
+	Name           string
+	Owner          string
+	Value          *sdcpb.TypedValue
+	DeviationValue *sdcpb.TypedValue
+	// KeyName is set when the level is the key level of a list.
+	KeyName  string
+	Children []*BlameNode
+}

pkg/core/core_test.go

@@ -20,14 +20,14 @@ func newTestConfig(t *testing.T) *config.Config {
 	cfg.SchemaStore = &schemaConfig.SchemaStoreConfig{
 		Type: schemaConfig.StoreTypeMemory,
 		Schemas: []*schemaConfig.SchemaConfig{{
-			Name: "s", Vendor: "v", Version: "1",
-			Files: []string{"../../tests/schema/sdcio_model_if.yang"},
+			Name: "", Vendor: "v", Version: "1",
+			Files: []string{"../../tests/schema/sdcio_model.yang"}, Directories: []string{"../../tests/schema"},
 		}},
 	}
 	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"},
+		Schema: &config.SchemaConfig{Name: "", Vendor: "v", Version: "1"},
 		SBI:    &config.SBI{Type: "noop"},
 		Sync:   &config.Sync{},
 	}}

pkg/core/datastore.go

@@ -23,27 +23,6 @@ func (c *Core) GetDataStore(ctx context.Context, name string) (*api.DatastoreInf
 	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 {

pkg/core/datastore_admin.go

@@ -0,0 +1,185 @@
+package core
+
+import (
+	"context"
+	"fmt"
+	"math"
+	"strings"
+	"time"
+
+	"github.com/sdcio/data-server/pkg/api"
+	"github.com/sdcio/data-server/pkg/config"
+	"github.com/sdcio/data-server/pkg/datastore"
+	logf "github.com/sdcio/logger"
+)
+
+// CreateDataStore creates and registers a datastore from spec. It fails with
+// api.ErrInvalidArgument for a missing or oversized name, an existing name, an
+// unknown target or sync protocol, or an invalid resulting config.
+func (c *Core) CreateDataStore(ctx context.Context, spec api.DatastoreSpec) error {
+	log := logf.FromContext(ctx).WithName("CreateDataStore").WithValues("datastore-name", spec.Name)
+
+	if spec.Name == "" {
+		return fmt.Errorf("%w: missing datastore name", api.ErrInvalidArgument)
+	}
+	if len(spec.Name) > math.MaxUint16 {
+		return fmt.Errorf("%w: datastore name too long", api.ErrInvalidArgument)
+	}
+	if _, err := c.getDatastore(spec.Name); err == nil {
+		return fmt.Errorf("%w: datastore %s already exists", api.ErrInvalidArgument, spec.Name)
+	}
+
+	dsCfg, err := c.datastoreConfig(spec)
+	if err != nil {
+		return err
+	}
+	if err := dsCfg.ValidateSetDefaults(); err != nil {
+		return fmt.Errorf("%w: invalid datastore config: %v", api.ErrInvalidArgument, err)
+	}
+
+	ds, err := datastore.New(c.ctx, dsCfg, c.schemaClient, c.cacheClient, c.dialOpts...)
+	if err != nil {
+		return err
+	}
+	if err := c.addDatastore(ds); err != nil {
+		_ = ds.Stop(ctx)
+		return fmt.Errorf("%w: %v", api.ErrInvalidArgument, err)
+	}
+	log.Info("datastore created successfully")
+	return nil
+}
+
+// DeleteDataStore stops the named datastore and removes it with its cached
+// state. It fails with api.ErrInvalidArgument for an empty name and
+// api.ErrUnknownDatastore when no such datastore exists.
+func (c *Core) DeleteDataStore(ctx context.Context, name string) error {
+	log := logf.FromContext(ctx).WithName("DeleteDataStore").WithValues("datastore-name", name)
+
+	c.mu.Lock()
+	ds, ok := c.datastores[name]
+	if name != "" && ok {
+		delete(c.datastores, name)
+	}
+	c.mu.Unlock()
+	if name == "" {
+		return fmt.Errorf("%w: missing datastore name", api.ErrInvalidArgument)
+	}
+	if !ok {
+		return fmt.Errorf("%w %s", api.ErrUnknownDatastore, name)
+	}
+
+	if err := ds.Stop(c.ctx); err != nil {
+		log.Error(err, "failed to stop datastore")
+	}
+	_ = ds.Delete(ctx)
+	log.Info("deleted datastore")
+	return nil
+}
+
+func (c *Core) datastoreConfig(spec api.DatastoreSpec) (*config.DatastoreConfig, error) {
+	sbi, err := sbiFromTarget(&spec.Target)
+	if err != nil {
+		return nil, err
+	}
+	dsCfg := &config.DatastoreConfig{
+		Name: spec.Name,
+		Schema: &config.SchemaConfig{
+			Vendor:  spec.Schema.Vendor,
+			Version: spec.Schema.Version,
+		},
+		SBI:        sbi,
+		Validation: c.cfg.Validation.DeepCopy(),
+		Deviation:  c.cfg.Deviation.DeepCopy(),
+	}
+	dsCfg.Sync = &config.Sync{
+		Config: make([]*config.SyncProtocol, 0, len(spec.Sync.Configs)),
+	}
+	for i := range spec.Sync.Configs {
+		sp, err := syncProtocol(&spec.Sync.Configs[i])
+		if err != nil {
+			return nil, err
+		}
+		dsCfg.Sync.Config = append(dsCfg.Sync.Config, sp)
+	}
+	return dsCfg, nil
+}
+
+func sbiFromTarget(t *api.Target) (*config.SBI, error) {
+	sbi := &config.SBI{Type: string(t.Type), Address: t.Address, Port: t.Port}
+	switch strings.ToLower(string(t.Type)) {
+	case "netconf":
+		o := t.ProtocolOptions.Netconf
+		if o == nil {
+			o = &api.NetconfOptions{}
+		}
+		commit := "candidate"
+		switch o.CommitCandidate {
+		case api.CommitCandidateCandidate:
+		case api.CommitCandidateRunning:
+			commit = "running"
+		default:
+			return nil, fmt.Errorf("%w: unknown commit datastore %v", api.ErrInvalidArgument, o.CommitCandidate)
+		}
+		sbi.NetconfOptions = &config.SBINetconfOptions{
+			IncludeNS:              o.IncludeNs,
+			OperationWithNamespace: o.OperationWithNs,
+			UseOperationRemove:     o.UseOperationRemove,
+			CommitDatastore:        commit,
+		}
+	case "gnmi":
+		g := t.ProtocolOptions.GNMI
+		if g == nil {
+			g = &api.GNMIOptions{}
+		}
+		sbi.GnmiOptions = &config.SBIGnmiOptions{Encoding: string(g.Encoding), TargetName: g.TargetName}
+	case "noop":
+	default:
+		return nil, fmt.Errorf("%w: unknown target connection protocol type %q", api.ErrInvalidArgument, t.Type)
+	}
+	if t.TLS != nil {
+		sbi.TLS = &config.TLS{CA: t.TLS.CA, Cert: t.TLS.Cert, Key: t.TLS.Key, SkipVerify: t.TLS.SkipVerify}
+	}
+	if t.Credentials != nil {
+		sbi.Credentials = &config.Creds{Username: t.Credentials.Username, Password: t.Credentials.Password, Token: t.Credentials.Token}
+	}
+	return sbi, nil
+}
+
+func syncProtocol(sc *api.SyncConfig) (*config.SyncProtocol, error) {
+	var typ string
+	var gnmi *api.GNMIOptions
+	if sc.Target != nil {
+		typ = string(sc.Target.Type)
+		gnmi = sc.Target.ProtocolOptions.GNMI
+	}
+	sp := &config.SyncProtocol{
+		Protocol: typ,
+		Name:     sc.Name,
+		Paths:    sc.Paths,
+		Interval: time.Duration(sc.Interval),
+	}
+	switch strings.ToLower(typ) {
+	case "gnmi":
+		sp.Mode = syncModeName(sc.Mode)
+		if gnmi != nil {
+			sp.Encoding = string(gnmi.Encoding)
+		}
+	case "netconf":
+	default:
+		return nil, fmt.Errorf("%w: unknown target sync protocol %q", api.ErrInvalidArgument, typ)
+	}
+	return sp, nil
+}
+
+func syncModeName(m api.SyncMode) string {
+	switch m {
+	case api.SyncModeSample:
+		return "sample"
+	case api.SyncModeOnce:
+		return "once"
+	case api.SyncModeGet:
+		return "get"
+	default:
+		return "on-change"
+	}
+}