[core] Apricot host-detector inventory cache

Author: teo

apricot/cacheproxy/service.go

@@ -0,0 +1,146 @@
+package cacheproxy
+
+import (
+	"github.com/AliceO2Group/Control/configuration"
+	"github.com/AliceO2Group/Control/configuration/componentcfg"
+)
+
+// Implements a cache proxy Service for the configuration system.
+// Only DetectorForHost/DetectorsForHosts are cached, all other calls are passed through.
+type Service struct{
+	base configuration.Service
+	cache svcCache
+}
+
+type svcCache struct {
+	detectorsInventory map[string][]string
+	detectorForHost map[string]string
+}
+
+func NewService(base configuration.Service) (*Service, error) {
+	svc := &Service{
+		base: base,
+		cache: svcCache{
+			detectorsInventory: make(map[string][]string),
+			detectorForHost: make(map[string]string),
+		},
+	}
+
+	var err error
+	svc.cache.detectorsInventory, err = svc.base.GetDetectorsInventory()
+	if err != nil {
+		return nil, err
+	}
+
+	for det, hosts := range svc.cache.detectorsInventory {
+		for _, host := range hosts {
+			svc.cache.detectorForHost[host] = det
+		}
+	}
+
+	return svc, nil
+}
+
+func (s Service) GetRuntimeEntry(component string, key string) (string, error) {
+	return s.base.GetRuntimeEntry(component, key)
+}
+
+func (s Service) SetRuntimeEntry(component string, key string, value string) error {
+	return s.base.SetRuntimeEntry(component, key, value)
+}
+
+func (s Service) ListRuntimeEntries(component string) ([]string, error) {
+	return s.base.ListRuntimeEntries(component)
+}
+
+func (s Service) NewRunNumber() (runNumber uint32, err error) {
+	return s.base.NewRunNumber()
+}
+
+func (s Service) GetDefaults() map[string]string {
+	return s.base.GetDefaults()
+}
+
+func (s Service) GetVars() map[string]string {
+	return s.base.GetVars()
+}
+
+func (s Service) GetComponentConfiguration(query *componentcfg.Query) (payload string, err error) {
+	return s.base.GetComponentConfiguration(query)
+}
+
+func (s Service) GetComponentConfigurationWithLastIndex(query *componentcfg.Query) (payload string, lastIndex uint64, err error) {
+	return s.base.GetComponentConfigurationWithLastIndex(query)
+}
+
+func (s Service) GetAndProcessComponentConfiguration(query *componentcfg.Query, varStack map[string]string) (payload string, err error) {
+	return s.base.GetAndProcessComponentConfiguration(query, varStack)
+}
+
+func (s Service) ListDetectors(getAll bool) (detectors []string, err error) {
+	return s.base.ListDetectors(getAll)
+}
+
+func (s Service) GetHostInventory(detector string) (hosts []string, err error) {
+	return s.base.GetHostInventory(detector)
+}
+
+func (s Service) GetDetectorsInventory() (inventory map[string][]string, err error) {
+	return s.base.GetDetectorsInventory()
+}
+
+func (s Service) ListComponents() (components []string, err error) {
+	return s.base.ListComponents()
+}
+
+func (s Service) ListComponentEntries(query *componentcfg.EntriesQuery, showLatestTimestamp bool) (entries []string, err error) {
+	return s.base.ListComponentEntries(query, showLatestTimestamp)
+}
+
+func (s Service) ListComponentEntryHistory(query *componentcfg.Query) (entries []string, err error) {
+	return s.base.ListComponentEntryHistory(query)
+}
+
+func (s Service) ResolveComponentQuery(query *componentcfg.Query) (resolved *componentcfg.Query, err error) {
+	return s.base.ResolveComponentQuery(query)
+}
+
+func (s Service) ImportComponentConfiguration(query *componentcfg.Query, payload string, newComponent bool, useVersioning bool) (existingComponentUpdated bool, existingEntryUpdated bool, newTimestamp int64, err error) {
+	return s.base.ImportComponentConfiguration(query, payload, newComponent, useVersioning)
+}
+
+func (s Service) GetDetectorForHost(hostname string) (string, error) {
+	det, ok := s.cache.detectorForHost[hostname]
+	if !ok {
+		return s.base.GetDetectorForHost(hostname)
+	}
+	return det, nil
+}
+
+func (s Service) GetDetectorsForHosts(hosts []string) ([]string, error) {
+	detectors := make(map[string]struct{}, 0)
+	for _, host := range hosts {
+		det, ok := s.cache.detectorForHost[host]
+		if !ok {
+			return s.base.GetDetectorsForHosts(hosts)
+		}
+		detectors[det] = struct{}{}
+	}
+	detList := make([]string, 0, len(detectors))
+	for det, _ := range detectors {
+		detList = append(detList, det)
+	}
+	return detList, nil
+}
+
+func (s Service) GetCRUCardsForHost(hostname string) (string, error) {
+	return s.base.GetCRUCardsForHost(hostname)
+}
+
+func (s Service) GetEndpointsForCRUCard(hostname, cardSerial string) (string, error) {
+	return s.base.GetEndpointsForCRUCard(hostname, cardSerial)
+}
+
+func (s Service) RawGetRecursive(path string) (string, error) {
+	return s.base.RawGetRecursive(path)
+}

apricot/instance.go

@@ -29,6 +29,7 @@ import (
 	"net/url"
 	"sync"
 
+	"github.com/AliceO2Group/Control/apricot/cacheproxy"
 	"github.com/AliceO2Group/Control/apricot/local"
 	"github.com/AliceO2Group/Control/apricot/remote"
 	"github.com/AliceO2Group/Control/common/logger/infologger"
@@ -47,6 +48,8 @@ func newService(configUri string) (configuration.Service, error) {
 		return nil, err
 	}
 
+	var svc configuration.Service
+
 	switch parsedUri.Scheme {
 	case "consul":
 		fallthrough
@@ -62,7 +65,14 @@ func newService(configUri string) (configuration.Service, error) {
 				WithField("level", infologger.IL_Support).
 				Info("new embedded apricot instance")
 		}
-		return local.NewService(configUri)
+		svc, err = local.NewService(configUri)
+		if err != nil {
+			return svc, err
+		}
+		if viper.GetBool("configCache") {
+			svc, err = cacheproxy.NewService(svc)
+		}
+		return svc, err
 	case "apricot":
 		if viper.GetString("component") == "apricot" {
 			log.WithField("configUri", configUri).
@@ -75,7 +85,14 @@ func newService(configUri string) (configuration.Service, error) {
 				WithField("level", infologger.IL_Support).
 				Info("new apricot client")
 		}
-		return remote.NewService(configUri)
+		svc, err = remote.NewService(configUri)
+		if err != nil {
+			return svc, err
+		}
+		if viper.GetBool("configCache") {
+			svc, err = cacheproxy.NewService(svc)
+		}
+		return svc, err
 	default:
 		return nil, fmt.Errorf("invalid configuration URI scheme %s", parsedUri.Scheme)
 	}

core/config.go

@@ -118,6 +118,7 @@ func setDefaults() error {
 	viper.SetDefault("concurrentWorkflowTemplateProcessing", false)
 	viper.SetDefault("concurrentIteratorRoleExpansion", false)
 	viper.SetDefault("reuseUnlockedTasks", false)
+	viper.SetDefault("configCache", true)
 	return nil
 }
 
@@ -179,6 +180,7 @@ func setFlags() error {
 	pflag.Bool("concurrentWorkflowTemplateProcessing", viper.GetBool("concurrentWorkflowTemplateProcessing"), "Process workflow templates concurrently")
 	pflag.Bool("concurrentIteratorRoleExpansion", viper.GetBool("concurrentIteratorRoleExpansion"), "Expand iterator roles concurrently during workflow template processing")
 	pflag.Bool("reuseUnlockedTasks", viper.GetBool("reuseUnlockedTasks"), "Reuse unlocked active tasks when satisfying environment deployment requests")
+	pflag.Bool("configCache", viper.GetBool("configCache"), "Enable cache layer between AliECS core and Apricot")
 
 	pflag.Parse()
 	return viper.BindPFlags(pflag.CommandLine)