Resolve "Make interval, amount of keys to poll from qkd modules and max keys configurable"

See merge request !180

README.md

@@ -89,6 +89,9 @@ Peers:
       Mastermode: true # (Type: etsi specific) Sets the QuantumModule in MasterMode. Keys are requested via GetKey and synced with peer KMS.
       LocalSAEID: "localSAEID" # the related ID of the local SAE
       TargetSAEID: "targetSAEID" # the related ID of the target SAE
+      KeyFetchInterval: 10 # interval in seconds for how often keys should be requested from a QuantumModule (optional, uses default if not provided)
+      KeyFetchAmount: 10 # amount of keys to be requested from a QuantumModule (optional, uses default if not provided)
+      MaxKeyFillLevel: 100 # maximum number of keys to be stored in the storage related to one KMS peer (should be the same for the peer on the other side, uses default if not provided)
   # peer to goKMS03
   - PeerId: "f80db2c0-2480-46b9-b7d1-b63f954e8227"
     PeerInterComAddr: 172.100.20.12:50910

goKMS/config/config.go

@@ -34,12 +34,15 @@ type TLSConfig struct {
 }
 
 type QuantumModule struct {
-	QmType      string `yaml:"Type"`
-	Address     string `yaml:"Address"`
-	Hostname    string `yaml:"Hostname"`
-	LocalSAEID  string `yaml:"LocalSAEID"`
-	TargetSAEID string `yaml:"TargetSAEID"`
-	MasterMode  bool   `yaml:"MasterMode"`
+	QmType           string `yaml:"Type"`
+	Address          string `yaml:"Address"`
+	Hostname         string `yaml:"Hostname"`
+	LocalSAEID       string `yaml:"LocalSAEID"`
+	TargetSAEID      string `yaml:"TargetSAEID"`
+	MasterMode       bool   `yaml:"MasterMode"`
+	KeyFetchInterval int    `yaml:"KeyFetchInterval"`
+	KeyFetchAmount   int    `yaml:"KeyFetchAmount"`
+	MaxKeyFillLevel  int    `yaml:"MaxKeyFillLevel"`
 }
 
 type ETSI14Server struct {

goKMS/gnmiHandlers/kms/keyStoreHandler.go

+package kmsHandler
+
+import (
+	"fmt"
+
+	"code.fbi.h-da.de/danet/gnmi-target/handler"
+	"code.fbi.h-da.de/danet/quant/goKMS/kms"
+	"code.fbi.h-da.de/danet/quant/goKMS/kms/event"
+	gnmitargetygot "code.fbi.h-da.de/danet/quant/goKMS/model"
+	"github.com/openconfig/gnmi/proto/gnmi"
+	"github.com/openconfig/ygot/ygot"
+	"github.com/sirupsen/logrus"
+)
+
+type KeyStoreHandler struct {
+	handler.DefaultPathHandler
+	kms                     *kms.KMS
+	maxKeyFillLevelsDefined map[string]uint64
+	events                  <-chan event.Event
+}
+
+type keyFillLevel struct {
+	storeID   string
+	fillLevel uint64
+}
+
+func NewKeyStoreHandler(kms *kms.KMS, maxKeyFillLevelsDefined map[string]uint64) *KeyStoreHandler {
+	return &KeyStoreHandler{
+		DefaultPathHandler: handler.DefaultPathHandler{
+			Name: "key-store-handler",
+			Paths: map[string]struct{}{
+				"key-stores": {},
+			},
+		},
+		kms:                     kms,
+		maxKeyFillLevelsDefined: maxKeyFillLevelsDefined,
+	}
+}
+
+func (yh *KeyStoreHandler) Init(config *handler.Config, publishToSubsFunc func([]*gnmi.Notification) error) error {
+	yh.Config = config
+	yh.PublishToSubs = publishToSubsFunc
+
+	var err error
+	yh.events, err = yh.kms.EventBus().Subscribe(event.KEY_STORE)
+	if err != nil {
+		return err
+	}
+
+	_, err = yh.updateOrCreateKeyStoreHandler()
+	if err != nil {
+		return err
+	}
+
+	// Start the go routine that takes care of any update from the kms
+	go func() {
+		for {
+			select {
+			case <-yh.events:
+				logrus.Println("Update for KeyStores.")
+				_, err := yh.updateOrCreateKeyStoreHandler()
+				if err != nil {
+					logrus.Errorf("Error within key stores subscription goroutine; %v", err)
+				}
+
+				// gnmi subscribe things here?
+			}
+		}
+	}()
+
+	return nil
+}
+
+func (yh *KeyStoreHandler) Update(c ygot.ValidatedGoStruct, jobs []*gnmi.Update) error {
+	return nil
+}
+
+func (yh *KeyStoreHandler) updateOrCreateKeyStoreHandler() ([]*gnmi.Notification, error) {
+	yh.Config.Lock()
+	defer yh.Config.Unlock()
+
+	copyCurrentConfig, err := ygot.DeepCopy(yh.Config.Data)
+	if err != nil {
+		return nil, err
+	}
+
+	newConfig, ok := copyCurrentConfig.(*gnmitargetygot.Gnmitarget)
+	if !ok {
+		return nil, fmt.Errorf("Wrong type, exptected: %T, got: %T", (*gnmitargetygot.Temp_KeyStores)(nil), copyCurrentConfig)
+	}
+
+	confKeyStores := newConfig.GetOrCreateKeyStores()
+
+	keyFillLevels := getKeyFillLevels(yh.kms)
+
+	// TODO: Maybe add more config values here!
+	for _, keyFillLevel := range keyFillLevels {
+		confKeyStoreContainer := confKeyStores.GetOrCreateKeyStore(keyFillLevel.storeID)
+
+		confKeyStoreContainer.KmsPeerId = ygot.String(keyFillLevel.storeID)
+		confKeyStore := confKeyStoreContainer.GetOrCreateKeyStore()
+
+		confKeyStore.KeyFillLevel = ygot.Uint64(keyFillLevel.fillLevel)
+		maxKeyFillLevel, ok := yh.maxKeyFillLevelsDefined[keyFillLevel.storeID]
+		if !ok {
+			return nil, fmt.Errorf("no max key fill level available for store with ID: %s", keyFillLevel.storeID) // TODO(faseid): check if really want to return here?!
+		}
+
+		confKeyStore.MaxKeyFillLevel = ygot.Uint64(maxKeyFillLevel)
+	}
+
+	// validate struct
+	if err := newConfig.Validate(); err != nil {
+		return nil, err
+	}
+
+	notifications, err := ygot.DiffWithAtomic(yh.Config.Data, newConfig)
+	if err != nil {
+		return nil, err
+	}
+
+	yh.Config.Data = newConfig
+
+	return notifications, nil
+}
+
+func getKeyFillLevels(kms *kms.KMS) []keyFillLevel {
+	kmsPeers := kms.KmsPeers
+	keyFillLevels := []keyFillLevel{}
+
+	for _, peer := range kmsPeers {
+		keyFillLevels = append(keyFillLevels, keyFillLevel{
+			storeID:   peer.GetKmsPeerId().String(),
+			fillLevel: uint64(peer.GetKeyStore().Length()),
+		})
+	}
+
+	return keyFillLevels
+}

goKMS/kms/event/event.go

@@ -9,6 +9,7 @@ const (
 	ROUTE
 	QUANTUM_MODULE
 	CREATE_ROUTE
+	KEY_STORE
 )
 
 // Event ...
@@ -86,3 +87,23 @@ func (e *RouteEvent) Topic() Topic {
 func (e *RouteEvent) Time() time.Time {
 	return e.Timestamp
 }
+
+type KeyStoresEvent struct {
+	EventTopic Topic
+	Timestamp  time.Time
+}
+
+func NewKeyStoresEvent() *KeyStoresEvent {
+	return &KeyStoresEvent{
+		EventTopic: KEY_STORE,
+		Timestamp:  time.Now(),
+	}
+}
+
+func (e *KeyStoresEvent) Topic() Topic {
+	return e.EventTopic
+}
+
+func (e *KeyStoresEvent) Time() time.Time {
+	return e.Timestamp
+}

goKMS/kms/kms.go

@@ -190,7 +190,10 @@ func (kms *KMS) initializePeers(config *config.Config) error {
 		case "emulated":
 			qm = peers.NewDanetQuantumModule(pqm.Address, config.Id)
 		case "etsi":
-			qm, err = peers.NewETSI014HTTPQuantumModule(pqm.Address, config.Id, pqm.LocalSAEID, pqm.TargetSAEID, config.QuantumModuleTLS, pqm.MasterMode)
+			qm, err = peers.NewETSI014HTTPQuantumModule(pqm.Address, config.Id, pqm.LocalSAEID, pqm.TargetSAEID,
+				config.QuantumModuleTLS, pqm.MasterMode,
+				peer.QuantumModule.KeyFetchInterval, int64(peer.QuantumModule.KeyFetchAmount), uint64(peer.QuantumModule.MaxKeyFillLevel),
+				kms.eventBus)
 			if err != nil {
 				log.Fatalf("Failed to create ETSI QKD module: %s", err)
 				return nil

goKMS/kms/peers/etsi14Quantummodule.go

@@ -18,19 +18,32 @@ import (
 	log "github.com/sirupsen/logrus"
 )
 
+const (
+	maxFailedKeyRequestAttempts = 10
+	defaultKeyFetchInterval     = 10
+	defaultKeyFetchAmount       = int64(1)
+	defualtMaxKeyFillLevel      = uint64(100)
+
+	backgroundKeyStoreUpdateInterval = 1
+)
+
 type ETSI014HTTPQuantumModule struct {
-	id          uuid.UUID
-	kmsId       string
-	addr        string
-	keyStore    *store.KmsKeyStore
-	kmsClient   *GRPCClient
-	client      *etsi14ClientImpl.ClientImpl
-	localSAEID  string
-	targetSAEID string
-	master      bool
+	id               uuid.UUID
+	kmsId            string
+	addr             string
+	keyStore         *store.KmsKeyStore
+	kmsClient        *GRPCClient
+	client           *etsi14ClientImpl.ClientImpl
+	localSAEID       string
+	targetSAEID      string
+	master           bool
+	keyFetchInterval int
+	keyFetchAmount   int64
+	maxKeyFillLevel  uint64
+	kmsEventBus      *event.EventBus
 }
 
-func NewETSI014HTTPQuantumModule(addr, kmsId, localSAEID, targetSAEID string, tlsConfig config.TLSConfig, master bool) (*ETSI014HTTPQuantumModule, error) {
+func NewETSI014HTTPQuantumModule(addr, kmsId, localSAEID, targetSAEID string, tlsConfig config.TLSConfig, master bool, keyFetchInterval int, keyFetchAmount int64, maxKeyFillLevel uint64, kmsEventBus *event.EventBus) (*ETSI014HTTPQuantumModule, error) {
 	parsedUrl, err := url.Parse(addr)
 	if err != nil {
 		return nil, err
@@ -65,16 +78,33 @@ func NewETSI014HTTPQuantumModule(addr, kmsId, localSAEID, targetSAEID string, tl
 		return nil, err
 	}
 
+	// set defaults for key fetching if not defined in config
+	if keyFetchInterval == 0 {
+		keyFetchInterval = defaultKeyFetchInterval
+	}
+
+	if keyFetchAmount == 0 {
+		keyFetchAmount = defaultKeyFetchAmount
+	}
+
+	if maxKeyFillLevel == 0 {
+		maxKeyFillLevel = defualtMaxKeyFillLevel
+	}
+
 	return &ETSI014HTTPQuantumModule{
-		id:          uuid.New(),
-		kmsId:       kmsId,
-		addr:        addr,
-		keyStore:    store.NewKmsKeyStore(256),
-		kmsClient:   nil,
-		client:      client,
-		localSAEID:  localSAEID,
-		targetSAEID: targetSAEID,
-		master:      master,
+		id:               uuid.New(),
+		kmsId:            kmsId,
+		addr:             addr,
+		keyStore:         store.NewKmsKeyStore(256),
+		kmsClient:        nil,
+		client:           client,
+		localSAEID:       localSAEID,
+		targetSAEID:      targetSAEID,
+		master:           master,
+		keyFetchInterval: keyFetchInterval,
+		keyFetchAmount:   keyFetchAmount,
+		maxKeyFillLevel:  maxKeyFillLevel,
+		kmsEventBus:      kmsEventBus,
 	}, nil
 }
 
@@ -87,38 +117,54 @@ func (qm *ETSI014HTTPQuantumModule) Client() *etsi14ClientImpl.ClientImpl {
 }
 
 func (qm *ETSI014HTTPQuantumModule) Initialize() error {
-	// start polling
+	// sends events on the event bus every x seconds to keep key store config updated
+	go qm.runBackgroundKeyStoreUpdates()
+
+	// start polling keys
 	if qm.master {
 		go func() {
-			ticker := time.NewTicker(2 * time.Second)
+			ticker := time.NewTicker(time.Duration(qm.keyFetchInterval) * time.Second)
 			defer ticker.Stop()
 
+			failedAttemps := 0
+
 			// TODO: add context/channel to stop
 			for range ticker.C {
-				container, err := qm.GetKeys(1, 256, nil, nil, nil)
-				if err != nil {
-					log.Error(err)
+				if failedAttemps == maxFailedKeyRequestAttempts {
+					log.Errorf("stopped trying to fetch keys from qkd module after %d tries", failedAttemps)
 					break
 				}
 
-				keyIds := make([]string, len(container.GetKeys()))
-				for i, keyItem := range container.GetKeys() {
-					keyIds[i] = keyItem.GetKeyID()
-				}
-
-				_, err = qm.kmsClient.KeyIdNotification(context.Background(),
-					&pbIC.KeyIdNotificationRequest{
-						Timestamp: time.Now().Unix(),
-						KmsId:     qm.kmsId,
-						KeyIds:    keyIds,
-					})
-				if err != nil {
-					log.Error(err)
-					break
-				}
-
-				if err := store.AddETSIKeysToKeystore(qm.keyStore, container.GetKeys()); err != nil {
-					log.Error(err)
+				if qm.keyStore.Length() < int(qm.maxKeyFillLevel) {
+					container, err := qm.GetKeys(qm.keyFetchAmount, 256, nil, nil, nil)
+					if err != nil {
+						log.Error(err)
+						failedAttemps++
+						continue
+					}
+
+					keyIds := make([]string, len(container.GetKeys()))
+					for i, keyItem := range container.GetKeys() {
+						keyIds[i] = keyItem.GetKeyID()
+					}
+
+					_, err = qm.kmsClient.KeyIdNotification(context.Background(),
+						&pbIC.KeyIdNotificationRequest{
+							Timestamp: time.Now().Unix(),
+							KmsId:     qm.kmsId,
+							KeyIds:    keyIds,
+						})
+					if err != nil {
+						log.Error(err)
+						failedAttemps++
+						continue
+					}
+
+					if err := store.AddETSIKeysToKeystore(qm.keyStore, container.GetKeys()); err != nil {
+						log.Error(err)
+					}
+
+					failedAttemps = 0
 				}
 			}
 		}()
@@ -179,3 +225,15 @@ func (qm *ETSI014HTTPQuantumModule) GetKeyWithIds(keyIds []etsi14ClientGenerated
 
 	return container, nil
 }
+
+func (qm *ETSI014HTTPQuantumModule) runBackgroundKeyStoreUpdates() {
+	ticker := time.NewTicker(backgroundKeyStoreUpdateInterval * time.Second)
+	defer ticker.Stop()
+
+	for range ticker.C {
+		err := qm.kmsEventBus.Publish(event.NewKeyStoresEvent())
+		if err != nil {
+			log.Error(err)
+		}
+	}
+}

goKMS/kms/peers/kmsPeer.go

@@ -10,6 +10,7 @@ import (
 	pbIC "code.fbi.h-da.de/danet/quant/goKMS/api/gen/proto/go/kmsintercom"
 	"code.fbi.h-da.de/danet/quant/goKMS/kms/crypto"
 	"code.fbi.h-da.de/danet/quant/goKMS/kms/event"
+	"code.fbi.h-da.de/danet/quant/goKMS/kms/store"
 	"code.fbi.h-da.de/danet/quant/goKMS/kms/util"
 	"github.com/google/uuid"
 	log "github.com/sirupsen/logrus"
@@ -214,3 +215,7 @@ func (kp *KmsPeer) SetStatus(updateStatus KmsPeerStatus) {
 func (kp *KmsPeer) GetKmsPeerId() uuid.UUID {
 	return kp.peerKmsId
 }
+
+func (kp *KmsPeer) GetKeyStore() *store.KmsKeyStore {
+	return kp.servingQuantumModul.KeyStore()
+}

goKMS/main.go

@@ -124,6 +124,8 @@ func main() {
 		log.Fatal(err)
 	}
 
+	maxKeyFillLevels := getMaxKeyFillLevelsFromConfig(kmsConfig.Peers)
+
 	// The registered path handlers sorted by priority. If specific
 	// handlers should be able to process their workload before others,
 	// then they should be placed in the front of the slice.
@@ -137,6 +139,7 @@ func main() {
 		kmsHandler.NewPeerHandler(kms),
 		kmsHandler.NewKeyRoutingSessionHandler(kms),
 		kmsHandler.NewAssignForwardingHandler(kms),
+		kmsHandler.NewKeyStoreHandler(kms, maxKeyFillLevels),
 	}
 
 	// The gnmiTarget implementation uses a flag to pass NO tls, so we have to invert our flag for it to work.
@@ -233,3 +236,13 @@ func setupQkdnManagerServer(kms *kms.KMS, config config.QkdnManagerServer) {
 
 	cancel()
 }
+
+func getMaxKeyFillLevelsFromConfig(peers []config.Peer) map[string]uint64 {
+	maxKeyFillLevels := make(map[string]uint64, 0)
+
+	for _, peer := range peers {
+		maxKeyFillLevels[peer.PeerId] = uint64(peer.QuantumModule.MaxKeyFillLevel)
+	}
+
+	return maxKeyFillLevels
+}