Newer
Older
/* Copyright 2017 Google Inc.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
https://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
// Package gnmiserver implements a gnmi server.
package gnmiserver
import (
"bytes"
"compress/gzip"
"encoding/json"
"fmt"
"io/ioutil"
"reflect"
"strconv"
"sync"
"time"
"golang.org/x/net/context"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
log "github.com/golang/glog"
"github.com/golang/protobuf/proto"
"github.com/openconfig/gnmi/value"
"github.com/openconfig/ygot/util"
"github.com/openconfig/ygot/ygot"
"github.com/openconfig/ygot/ytypes"
Malte Bauch
committed
"code.fbi.h-da.de/danet/gnmi-target/handler"
not "code.fbi.h-da.de/danet/gnmi-target/internal/notifications"
dpb "github.com/golang/protobuf/protoc-gen-go/descriptor"
pb "github.com/openconfig/gnmi/proto/gnmi"
)
// ConfigCallback is the signature of the function to apply a validated config to the physical device.
Malte Bauch
committed
//type ConfigCallback func(ygot.ValidatedGoStruct, ygot.ValidatedGoStruct) error
supportedEncodings = []pb.Encoding{pb.Encoding_JSON, pb.Encoding_JSON_IETF}
)
// Server struct maintains the data structure for device config and implements the interface of gnmi server. It supports Capabilities, Get, and Set APIs.
// Typical usage:
// g := grpc.NewServer()
// s, err := Server.NewServer(model, config, callback)
// pb.NewServer(g, s)
// reflection.Register(g)
// listen, err := net.Listen("tcp", ":8080")
// g.Serve(listen)
//
// For a real device, apply the config changes to the hardware in the callback function.
// Arguments:
//
// newConfig: new root config to be applied on the device.
//
// func callback(newConfig ygot.ValidatedGoStruct) error {
// // Apply the config to your device and return nil if success. return error if fails.
// //
// // Do something ...
// }
Malte Bauch
committed
model *Model
config ygot.ValidatedGoStruct
mu sync.RWMutex // mu is the RW lock to protect the access to config
YangModelChangeDispatcher *not.Dispatcher
Malte Bauch
committed
handlers []handler.PathHandler
}
// NewServer creates an instance of Server with given json config.
Malte Bauch
committed
func NewServer(model *Model, config ygot.ValidatedGoStruct, notifications *not.Dispatcher, handlers ...handler.PathHandler) (*Server, error) {
/*rootStruct, err := model.NewConfigStruct(config)
model: model,
config: config,
YangModelChangeDispatcher: notifications,
Malte Bauch
committed
handlers: handlers,
/*if config != nil && s.callback != nil {
if err := s.callback(config); err != nil {
Malte Bauch
committed
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
// callback is used to apply a validated new config to the physical device. The
// changed values are then sent to YangHandlers to apply the new config values
// to the physical device.
func (s *Server) callback(newConfig ygot.ValidatedGoStruct, existingConf ygot.ValidatedGoStruct) error {
// All applied successfully, so time for finding the diff and report this
// Generate gnmi notifications for subscribe
configDiff, err := ygot.DiffWithAtomic(existingConf, newConfig)
if err != nil {
//TODO error logging as we cannot do anything in this case
}
// NOTE: This is a test implementation and not particularly sophisticated,
// but for now the general idea is this: For every registered module we
// check if a changed value is one that the registered module is
// responsible for. If this is the case, then the handlers update method is
// called.
// The update method of a handler will receive the new config with all the
// changes and from there on the handler updates the given values within
// the os.
for _, handler := range s.handlers {
handlerJobs, err := checkHandlerPaths(handler, configDiff)
if err != nil {
return err
}
if len(handlerJobs) != 0 {
for _, handlerJob := range handlerJobs {
// NOTE: add delete
//if err := handler.Delete(newConfig, handlerJob.Updates); err != nil {
// return err
//}
if err := handler.Update(newConfig, handlerJob.Updates); err != nil {
return err
}
}
}
}
//NOTE: There shoud be a better place for this.
// Run through configDiff and generate Publish events...
for _, specificDiff := range configDiff {
// First for gnmi Updates
updates := specificDiff.GetUpdate()
for _, specificUpdate := range updates {
pathString, _ := ygot.PathToString(specificUpdate.Path)
log.Infof("specificDiff update %s with value of %s", pathString, specificUpdate.Val.String())
// Wrap Update into a notification and ship it off
updateNotification := createUpdateNotification(specificUpdate)
s.YangModelChangeDispatcher.Publish(pathString, updateNotification)
}
// Second for gnmi Deletes
deletes := specificDiff.GetDelete()
for _, specificDelete := range deletes {
pathString, _ := ygot.PathToString(specificDelete)
log.Infof("specificDiff delete %s ", pathString)
// Wrap Update into a notification and ship it off
updateNotification := createDeleteNotification(specificDelete)
s.YangModelChangeDispatcher.Publish(pathString, updateNotification)
}
}
return nil
}
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
// checkEncodingAndModel checks whether encoding and models are supported by the server. Return error if anything is unsupported.
func (s *Server) checkEncodingAndModel(encoding pb.Encoding, models []*pb.ModelData) error {
hasSupportedEncoding := false
for _, supportedEncoding := range supportedEncodings {
if encoding == supportedEncoding {
hasSupportedEncoding = true
break
}
}
if !hasSupportedEncoding {
return fmt.Errorf("unsupported encoding: %s", pb.Encoding_name[int32(encoding)])
}
for _, m := range models {
isSupported := false
for _, supportedModel := range s.model.modelData {
if reflect.DeepEqual(m, supportedModel) {
isSupported = true
break
}
}
if !isSupported {
return fmt.Errorf("unsupported model: %v", m)
}
}
return nil
}
// doDelete deletes the path from the json tree if the path exists. If success,
// it calls the callback function to apply the change to the device hardware.
func (s *Server) doDelete(jsonTree map[string]interface{}, prefix, path *pb.Path) (*pb.UpdateResult, error) {
// Update json tree of the device config
var curNode interface{} = jsonTree
pathDeleted := false
fullPath := gnmiFullPath(prefix, path)
schema := s.model.schemaTreeRoot
for i, elem := range fullPath.Elem { // Delete sub-tree or leaf node.
node, ok := curNode.(map[string]interface{})
if !ok {
break
}
// Delete node
if i == len(fullPath.Elem)-1 {
if elem.GetKey() == nil {
delete(node, elem.Name)
pathDeleted = true
break
}
pathDeleted = deleteKeyedListEntry(node, elem)
break
}
if curNode, schema = getChildNode(node, schema, elem, false); curNode == nil {
break
}
}
if reflect.DeepEqual(fullPath, pbRootPath) { // Delete root
for k := range jsonTree {
delete(jsonTree, k)
}
}
//// Apply the validated operation to the config tree and device.
// newConfig, err := s.toGoStruct(jsonTree)
// if err != nil {
// return nil, status.Error(codes.Internal, err.Error())
// }
// if s.callback != nil {
// if applyErr := s.callback(newConfig); applyErr != nil {
// if rollbackErr := s.callback(s.config); rollbackErr != nil {
// return nil, status.Errorf(codes.Internal, "error in rollback the failed operation (%v): %v", applyErr, rollbackErr)
// }
// return nil, status.Errorf(codes.Aborted, "error in applying operation to device: %v", applyErr)
// }
// }
}
return &pb.UpdateResult{
Path: path,
Op: pb.UpdateResult_DELETE,
}, nil
}
// doReplaceOrUpdate validates the replace or update operation to be applied to
// the device, modifies the json tree of the config struct, then calls the
// callback function to apply the operation to the device hardware.
func (s *Server) doReplaceOrUpdate(jsonTree map[string]interface{}, op pb.UpdateResult_Operation, prefix, path *pb.Path, val *pb.TypedValue) (*pb.UpdateResult, error) {
// Validate the operation.
fullPath := gnmiFullPath(prefix, path)
Malte Bauch
committed
emptyNode, entry, err := ytypes.GetOrCreateNode(s.model.schemaTreeRoot, s.model.newRootValue(), fullPath)
if err != nil {
return nil, status.Errorf(codes.NotFound, "path %v is not found in the config structure: %v", fullPath, err)
}
Malte Bauch
committed
if entry.ReadOnly() {
return nil, status.Errorf(codes.Internal, "It is not possible to change the value of a config false field at path: %s", entry.Path())
}
jsonIETFValue := val.GetJsonIetfVal()
var nodeVal interface{}
nodeStruct, ok := emptyNode.(ygot.ValidatedGoStruct)
if ok {
if err := s.model.jsonUnmarshaler(jsonIETFValue, nodeStruct); err != nil {
return nil, status.Errorf(codes.InvalidArgument, "unmarshaling json data to config struct fails: %v", err)
}
if err := nodeStruct.Validate(); err != nil {
return nil, status.Errorf(codes.InvalidArgument, "config data validation fails: %v", err)
}
var err error
if nodeVal, err = ygot.ConstructIETFJSON(nodeStruct, &ygot.RFC7951JSONConfig{}); err != nil {
msg := fmt.Sprintf("error in constructing IETF JSON tree from config struct: %v", err)
log.Error(msg)
return nil, status.Error(codes.Internal, msg)
}
} else {
var err error
if nodeVal, err = value.ToScalar(val); err != nil {
return nil, status.Errorf(codes.Internal, "cannot convert leaf node to scalar type: %v", err)
}
}
Malte Bauch
committed
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
// Check if val contains config false fields and return an error if thats
// the case.
if nodeStruct != nil {
nodeStructCopy, err := ygot.DeepCopy(nodeStruct)
if err != nil {
return nil, status.Errorf(codes.Internal, "could not create a deep copy of current nodeStruct: %v", err)
}
if err := ygot.PruneConfigFalse(entry, nodeStruct); err != nil {
return nil, status.Errorf(codes.Internal, "failed prune config false for nodeStruct: %v", err)
}
diff, err := ygot.Diff(nodeStructCopy, nodeStruct)
if err != nil {
return nil, status.Errorf(codes.Internal, "failed to create a diff for nodeStruct: %v", err)
}
if diff != nil && len(diff.Delete) != 0 {
paths := make([]string, 0)
for _, path := range diff.Delete {
p, err := ygot.PathToString(path)
if err != nil {
return nil, status.Errorf(codes.Internal, "failed to create string from path: %v", err)
}
paths = append(paths, p)
}
return nil, status.Errorf(codes.InvalidArgument, "Can not set config false paths: %v", paths)
}
} else if entry.ReadOnly() {
return nil, status.Errorf(codes.Internal, "It is not possible to change the value of a config false field at path: %s", entry.Path())
}
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
// Update json tree of the device config.
var curNode interface{} = jsonTree
schema := s.model.schemaTreeRoot
for i, elem := range fullPath.Elem {
switch node := curNode.(type) {
case map[string]interface{}:
// Set node value.
if i == len(fullPath.Elem)-1 {
if elem.GetKey() == nil {
if grpcStatusError := setPathWithoutAttribute(op, node, elem, nodeVal); grpcStatusError != nil {
return nil, grpcStatusError
}
break
}
if grpcStatusError := setPathWithAttribute(op, node, elem, nodeVal); grpcStatusError != nil {
return nil, grpcStatusError
}
break
}
if curNode, schema = getChildNode(node, schema, elem, true); curNode == nil {
return nil, status.Errorf(codes.NotFound, "path elem not found: %v", elem)
}
case []interface{}:
return nil, status.Errorf(codes.NotFound, "incompatible path elem: %v", elem)
default:
return nil, status.Errorf(codes.Internal, "wrong node type: %T", curNode)
}
}
if reflect.DeepEqual(fullPath, pbRootPath) { // Replace/Update root.
if op == pb.UpdateResult_UPDATE {
return nil, status.Error(codes.Unimplemented, "update the root of config tree is unsupported")
}
nodeValAsTree, ok := nodeVal.(map[string]interface{})
if !ok {
return nil, status.Errorf(codes.InvalidArgument, "expect a tree to replace the root, got a scalar value: %T", nodeVal)
}
for k := range jsonTree {
delete(jsonTree, k)
}
for k, v := range nodeValAsTree {
jsonTree[k] = v
}
}
//newConfig, err := s.toGoStruct(jsonTree)
//if err != nil {
// return nil, status.Error(codes.Internal, err.Error())
//}
//// Apply the validated operation to the device.
//if s.callback != nil {
// if applyErr := s.callback(newConfig); applyErr != nil {
// if rollbackErr := s.callback(s.config); rollbackErr != nil {
// return nil, status.Errorf(codes.Internal, "error in rollback the failed operation (%v): %v", applyErr, rollbackErr)
// }
// return nil, status.Errorf(codes.Aborted, "error in applying operation to device: %v", applyErr)
// }
//}
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
return &pb.UpdateResult{
Path: path,
Op: op,
}, nil
}
func (s *Server) toGoStruct(jsonTree map[string]interface{}) (ygot.ValidatedGoStruct, error) {
jsonDump, err := json.Marshal(jsonTree)
if err != nil {
return nil, fmt.Errorf("error in marshaling IETF JSON tree to bytes: %v", err)
}
goStruct, err := s.model.NewConfigStruct(jsonDump)
if err != nil {
return nil, fmt.Errorf("error in creating config struct from IETF JSON data: %v", err)
}
return goStruct, nil
}
// getGNMIServiceVersion returns a pointer to the gNMI service version string.
// The method is non-trivial because of the way it is defined in the proto file.
func getGNMIServiceVersion() (*string, error) {
gzB, _ := (&pb.Update{}).Descriptor()
r, err := gzip.NewReader(bytes.NewReader(gzB))
if err != nil {
return nil, fmt.Errorf("error in initializing gzip reader: %v", err)
}
defer r.Close()
b, err := ioutil.ReadAll(r)
if err != nil {
return nil, fmt.Errorf("error in reading gzip data: %v", err)
}
desc := &dpb.FileDescriptorProto{}
if err := proto.Unmarshal(b, desc); err != nil {
return nil, fmt.Errorf("error in unmarshaling proto: %v", err)
}
ver, err := proto.GetExtension(desc.Options, pb.E_GnmiService)
if err != nil {
return nil, fmt.Errorf("error in getting version from proto extension: %v", err)
}
return ver.(*string), nil
}
// deleteKeyedListEntry deletes the keyed list entry from node that matches the
// path elem. If the entry is the only one in keyed list, deletes the entire
// list. If the entry is found and deleted, the function returns true. If it is
// not found, the function returns false.
func deleteKeyedListEntry(node map[string]interface{}, elem *pb.PathElem) bool {
curNode, ok := node[elem.Name]
if !ok {
return false
}
keyedList, ok := curNode.([]interface{})
if !ok {
return false
}
for i, n := range keyedList {
m, ok := n.(map[string]interface{})
if !ok {
log.Errorf("expect map[string]interface{} for a keyed list entry, got %T", n)
return false
}
keyMatching := true
for k, v := range elem.Key {
attrVal, ok := m[k]
if !ok {
return false
}
if v != fmt.Sprintf("%v", attrVal) {
keyMatching = false
break
}
}
if keyMatching {
listLen := len(keyedList)
if listLen == 1 {
delete(node, elem.Name)
return true
}
keyedList[i] = keyedList[listLen-1]
node[elem.Name] = keyedList[0 : listLen-1]
return true
}
}
return false
}
// gnmiFullPath builds the full path from the prefix and path.
func gnmiFullPath(prefix, path *pb.Path) *pb.Path {
fullPath := &pb.Path{Origin: path.Origin}
if path.GetElement() != nil {
fullPath.Element = append(prefix.GetElement(), path.GetElement()...)
}
if path.GetElem() != nil {
fullPath.Elem = append(prefix.GetElem(), path.GetElem()...)
}
return fullPath
}
// isNIl checks if an interface is nil or its value is nil.
func isNil(i interface{}) bool {
if i == nil {
return true
}
switch kind := reflect.ValueOf(i).Kind(); kind {
case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Ptr, reflect.Slice:
return reflect.ValueOf(i).IsNil()
default:
return false
}
}
// setPathWithAttribute replaces or updates a child node of curNode in the IETF
// JSON config tree, where the child node is indexed by pathElem with attribute.
// The function returns grpc status error if unsuccessful.
func setPathWithAttribute(op pb.UpdateResult_Operation, curNode map[string]interface{}, pathElem *pb.PathElem, nodeVal interface{}) error {
nodeValAsTree, ok := nodeVal.(map[string]interface{})
if !ok {
return status.Errorf(codes.InvalidArgument, "expect nodeVal is a json node of map[string]interface{}, received %T", nodeVal)
}
m := getKeyedListEntry(curNode, pathElem, true)
if m == nil {
return status.Errorf(codes.NotFound, "path elem not found: %v", pathElem)
}
if op == pb.UpdateResult_REPLACE {
for k := range m {
delete(m, k)
}
}
for attrKey, attrVal := range pathElem.GetKey() {
m[attrKey] = attrVal
if asNum, err := strconv.ParseFloat(attrVal, 64); err == nil {
m[attrKey] = asNum
}
for k, v := range nodeValAsTree {
if k == attrKey && fmt.Sprintf("%v", v) != attrVal {
return status.Errorf(codes.InvalidArgument, "invalid config data: %v is a path attribute", k)
}
}
}
for k, v := range nodeValAsTree {
m[k] = v
}
return nil
}
// setPathWithoutAttribute replaces or updates a child node of curNode in the
// IETF config tree, where the child node is indexed by pathElem without
// attribute. The function returns grpc status error if unsuccessful.
func setPathWithoutAttribute(op pb.UpdateResult_Operation, curNode map[string]interface{}, pathElem *pb.PathElem, nodeVal interface{}) error {
target, hasElem := curNode[pathElem.Name]
nodeValAsTree, nodeValIsTree := nodeVal.(map[string]interface{})
if op == pb.UpdateResult_REPLACE || !hasElem || !nodeValIsTree {
curNode[pathElem.Name] = nodeVal
return nil
}
targetAsTree, ok := target.(map[string]interface{})
if !ok {
return status.Errorf(codes.Internal, "error in setting path: expect map[string]interface{} to update, got %T", target)
}
for k, v := range nodeValAsTree {
targetAsTree[k] = v
}
return nil
}
// Capabilities returns supported encodings and supported models.
func (s *Server) Capabilities(ctx context.Context, req *pb.CapabilityRequest) (*pb.CapabilityResponse, error) {
ver, err := getGNMIServiceVersion()
if err != nil {
return nil, status.Errorf(codes.Internal, "error in getting gnmi service version: %v", err)
}
return &pb.CapabilityResponse{
SupportedModels: s.model.modelData,
SupportedEncodings: supportedEncodings,
GNMIVersion: *ver,
}, nil
}
// Get implements the Get RPC in gNMI spec.
func (s *Server) Get(ctx context.Context, req *pb.GetRequest) (*pb.GetResponse, error) {
if req.GetType() != pb.GetRequest_ALL {
return nil, status.Errorf(codes.Unimplemented, "unsupported request type: %s", pb.GetRequest_DataType_name[int32(req.GetType())])
}
if err := s.checkEncodingAndModel(req.GetEncoding(), req.GetUseModels()); err != nil {
return nil, status.Error(codes.Unimplemented, err.Error())
}
prefix := req.GetPrefix()
paths := req.GetPath()
notifications := make([]*pb.Notification, len(paths))
s.mu.RLock()
defer s.mu.RUnlock()
for i, path := range paths {
// Get schema node for path from config struct.
fullPath := path
if prefix != nil {
fullPath = gnmiFullPath(prefix, path)
}
if fullPath.GetElem() == nil && fullPath.GetElement() != nil {
return nil, status.Error(codes.Unimplemented, "deprecated path element type is unsupported")
}
nodes, err := ytypes.GetNode(s.model.schemaTreeRoot, s.config, fullPath)
if len(nodes) == 0 || err != nil || util.IsValueNil(nodes[0].Data) {
return nil, status.Errorf(codes.NotFound, "path %v not found: %v", fullPath, err)
}
node := nodes[0].Data
ts := time.Now().UnixNano()
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
// Return leaf node.
if !ok {
var val *pb.TypedValue
switch kind := reflect.ValueOf(node).Kind(); kind {
case reflect.Ptr, reflect.Interface:
var err error
val, err = value.FromScalar(reflect.ValueOf(node).Elem().Interface())
if err != nil {
msg := fmt.Sprintf("leaf node %v does not contain a scalar type value: %v", path, err)
log.Error(msg)
return nil, status.Error(codes.Internal, msg)
}
case reflect.Int64:
enumMap, ok := s.model.enumData[reflect.TypeOf(node).Name()]
if !ok {
return nil, status.Error(codes.Internal, "not a GoStruct enumeration type")
}
val = &pb.TypedValue{
Value: &pb.TypedValue_StringVal{
StringVal: enumMap[reflect.ValueOf(node).Int()].Name,
},
}
default:
return nil, status.Errorf(codes.Internal, "unexpected kind of leaf node type: %v %v", node, kind)
}
update := &pb.Update{Path: path, Val: val}
notifications[i] = &pb.Notification{
Timestamp: ts,
Prefix: prefix,
Update: []*pb.Update{update},
}
continue
}
if req.GetUseModels() != nil {
return nil, status.Errorf(codes.Unimplemented, "filtering Get using use_models is unsupported, got: %v", req.GetUseModels())
}
// Return IETF JSON by default.
jsonEncoder := func() (map[string]interface{}, error) {
return ygot.ConstructIETFJSON(nodeStruct, &ygot.RFC7951JSONConfig{AppendModuleName: true})
}
jsonType := "IETF"
buildUpdate := func(b []byte) *pb.Update {
return &pb.Update{Path: path, Val: &pb.TypedValue{Value: &pb.TypedValue_JsonIetfVal{JsonIetfVal: b}}}
}
if req.GetEncoding() == pb.Encoding_JSON {
jsonEncoder = func() (map[string]interface{}, error) {
return ygot.ConstructInternalJSON(nodeStruct)
}
jsonType = "Internal"
buildUpdate = func(b []byte) *pb.Update {
return &pb.Update{Path: path, Val: &pb.TypedValue{Value: &pb.TypedValue_JsonVal{JsonVal: b}}}
}
}
jsonTree, err := jsonEncoder()
if err != nil {
msg := fmt.Sprintf("error in constructing %s JSON tree from requested node: %v", jsonType, err)
log.Error(msg)
return nil, status.Error(codes.Internal, msg)
}
jsonDump, err := json.Marshal(jsonTree)
if err != nil {
msg := fmt.Sprintf("error in marshaling %s JSON tree to bytes: %v", jsonType, err)
log.Error(msg)
return nil, status.Error(codes.Internal, msg)
}
update := buildUpdate(jsonDump)
notifications[i] = &pb.Notification{
Timestamp: ts,
Prefix: prefix,
Update: []*pb.Update{update},
}
}
return &pb.GetResponse{Notification: notifications}, nil
}
// Set implements the Set RPC in gNMI spec.
func (s *Server) Set(ctx context.Context, req *pb.SetRequest) (*pb.SetResponse, error) {
s.mu.Lock()
defer s.mu.Unlock()
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
jsonTree, err := ygot.ConstructIETFJSON(s.config, &ygot.RFC7951JSONConfig{})
if err != nil {
msg := fmt.Sprintf("error in constructing IETF JSON tree from config struct: %v", err)
log.Error(msg)
return nil, status.Error(codes.Internal, msg)
}
prefix := req.GetPrefix()
var results []*pb.UpdateResult
for _, path := range req.GetDelete() {
res, grpcStatusError := s.doDelete(jsonTree, prefix, path)
if grpcStatusError != nil {
return nil, grpcStatusError
}
results = append(results, res)
}
for _, upd := range req.GetReplace() {
res, grpcStatusError := s.doReplaceOrUpdate(jsonTree, pb.UpdateResult_REPLACE, prefix, upd.GetPath(), upd.GetVal())
if grpcStatusError != nil {
return nil, grpcStatusError
}
results = append(results, res)
}
for _, upd := range req.GetUpdate() {
res, grpcStatusError := s.doReplaceOrUpdate(jsonTree, pb.UpdateResult_UPDATE, prefix, upd.GetPath(), upd.GetVal())
if grpcStatusError != nil {
return nil, grpcStatusError
}
results = append(results, res)
}
newConfig, err := s.toGoStruct(jsonTree)
if err != nil {
return nil, status.Error(codes.Internal, err.Error())
}
Malte Bauch
committed
currentConfig, err := ygot.DeepCopy(s.config)
if err != nil {
return nil, status.Error(codes.Internal, err.Error())
}
// Apply the validated operations to the device.
if s.callback != nil {
Malte Bauch
committed
if applyErr := s.callback(newConfig, currentConfig.(ygot.ValidatedGoStruct)); applyErr != nil {
if rollbackErr := s.callback(currentConfig.(ygot.ValidatedGoStruct), s.config); rollbackErr != nil {
return nil, status.Errorf(codes.Internal, "error in rollback the failed operation (%v): %v", applyErr, rollbackErr)
}
return nil, status.Errorf(codes.Aborted, "error in applying operation to device: %v", applyErr)
}
}
jsonDump, err := json.Marshal(jsonTree)
if err != nil {
msg := fmt.Sprintf("error in marshaling IETF JSON tree to bytes: %v", err)
log.Error(msg)
return nil, status.Error(codes.Internal, msg)
}
rootStruct, err := s.model.NewConfigStruct(jsonDump)
if err != nil {
msg := fmt.Sprintf("error in creating config struct from IETF JSON data: %v", err)
log.Error(msg)
return nil, status.Error(codes.Internal, msg)
}
s.config = rootStruct
return &pb.SetResponse{
Prefix: req.GetPrefix(),
Response: results,
Timestamp: time.Now().UnixNano(),
}, nil
}
// InternalUpdate is an experimental feature to let the server update its
// internal states. Use it with your own risk.
func (s *Server) InternalUpdate(fp func(config *ygot.ValidatedGoStruct) error) error {