diff options
Diffstat (limited to 'src/k8splugin/internal/rb')
| -rw-r--r-- | src/k8splugin/internal/rb/config.go | 436 | ||||
| -rw-r--r-- | src/k8splugin/internal/rb/config_backend.go | 439 | ||||
| -rw-r--r-- | src/k8splugin/internal/rb/config_template.go | 257 | ||||
| -rw-r--r-- | src/k8splugin/internal/rb/config_test.go | 259 |
4 files changed, 1391 insertions, 0 deletions
diff --git a/src/k8splugin/internal/rb/config.go b/src/k8splugin/internal/rb/config.go new file mode 100644 index 00000000..3bd8347b --- /dev/null +++ b/src/k8splugin/internal/rb/config.go @@ -0,0 +1,436 @@ +/* + * Copyright 2018 Intel Corporation, 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 + * + * http://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 rb + +import ( + pkgerrors "github.com/pkg/errors" + "k8splugin/internal/db" + "strconv" + "strings" +) + +// Config contains the parameters needed for configuration +type Config struct { + ConfigName string `json:"config-name"` + TemplateName string `json:"template-name"` + Description string `json:"description"` + Values map[string]interface{} `json:"values"` +} + +//ConfigResult output for Create, Update and delete +type ConfigResult struct { + DefinitionName string `json:"rb-name"` + DefinitionVersion string `json:"rb-version"` + ProfileName string `json:"profile-name"` + ConfigName string `json:"config-name"` + TemplateName string `json:"template-name"` + ConfigVersion uint `json:"config-verion"` +} + +//ConfigRollback input +type ConfigRollback struct { + AnyOf struct { + ConfigVersion string `json:"config-version,omitempty"` + ConfigTag string `json:"config-tag,omitempty"` + } `json:"anyOf"` +} + +//ConfigTagit for Tagging configurations +type ConfigTagit struct { + TagName string `json:"tag-name"` +} + +// ConfigManager is an interface exposes the config functionality +type ConfigManager interface { + Create(rbName, rbVersion, profileName string, p Config) (ConfigResult, error) + Get(rbName, rbVersion, profileName, configName string) (Config, error) + Help() map[string]string + Update(rbName, rbVersion, profileName, configName string, p Config) (ConfigResult, error) + Delete(rbName, rbVersion, profileName, configName string) (ConfigResult, error) + Rollback(rbName, rbVersion, profileName string, p ConfigRollback) error + Tagit(rbName, rbVersion, profileName string, p ConfigTagit) error +} + +// ConfigClient implements the ConfigManager +// It will also be used to maintain some localized state +type ConfigClient struct { + tagTag string +} + +// NewConfigClient returns an instance of the ConfigClient +// which implements the ConfigManager +func NewConfigClient() *ConfigClient { + return &ConfigClient{ + tagTag: "tag", + } +} + +// Help returns some information on how to create the content +// for the config in the form of html formatted page +func (v *ConfigClient) Help() map[string]string { + ret := make(map[string]string) + + return ret +} + +// Create an entry for the config in the database +func (v *ConfigClient) Create(rbName, rbVersion, profileName string, p Config) (ConfigResult, error) { + + // Check required fields + if p.ConfigName == "" || p.TemplateName == "" || len(p.Values) == 0 { + return ConfigResult{}, pkgerrors.New("Incomplete Configuration Provided") + } + cs := ConfigStore{ + rbName: rbName, + rbVersion: rbVersion, + profileName: profileName, + configName: p.ConfigName, + } + _, err := cs.getConfig() + if err == nil { + return ConfigResult{}, pkgerrors.Wrap(err, "Create Error - Config exists") + } else { + if strings.Contains(err.Error(), "Key doesn't exist") == false { + return ConfigResult{}, pkgerrors.Wrap(err, "Create Error") + } + } + lock, profileChannel := getProfileData(rbName + rbVersion + profileName) + // Acquire per profile Mutex + lock.Lock() + defer lock.Unlock() + err = applyConfig(rbName, rbVersion, profileName, p, profileChannel, "POST") + if err != nil { + return ConfigResult{}, pkgerrors.Wrap(err, "Apply Config failed") + } + // Create Config DB Entry + err = cs.createConfig(p) + if err != nil { + return ConfigResult{}, pkgerrors.Wrap(err, "Create Config DB Entry") + } + // Create Version Entry in DB for Config + cvs := ConfigVersionStore{ + rbName: rbName, + rbVersion: rbVersion, + profileName: profileName, + } + version, err := cvs.createConfigVersion(p, Config{}, "POST") + if err != nil { + return ConfigResult{}, pkgerrors.Wrap(err, "Create Config Version DB Entry") + } + // Create Result structure + cfgRes := ConfigResult{ + DefinitionName: rbName, + DefinitionVersion: rbVersion, + ProfileName: profileName, + ConfigName: p.ConfigName, + TemplateName: p.TemplateName, + ConfigVersion: version, + } + return cfgRes, nil +} + +// Update an entry for the config in the database +func (v *ConfigClient) Update(rbName, rbVersion, profileName, configName string, p Config) (ConfigResult, error) { + + // Check required fields + if len(p.Values) == 0 { + return ConfigResult{}, pkgerrors.New("Incomplete Configuration Provided") + } + // Check if Config exists + cs := ConfigStore{ + rbName: rbName, + rbVersion: rbVersion, + profileName: profileName, + configName: configName, + } + _, err := cs.getConfig() + if err != nil { + return ConfigResult{}, pkgerrors.Wrap(err, "Update Error - Config doesn't exist") + } + lock, profileChannel := getProfileData(rbName + rbVersion + profileName) + // Acquire per profile Mutex + lock.Lock() + defer lock.Unlock() + err = applyConfig(rbName, rbVersion, profileName, p, profileChannel, "PUT") + if err != nil { + return ConfigResult{}, pkgerrors.Wrap(err, "Apply Config failed") + } + // Update Config DB Entry + configPrev, err := cs.updateConfig(p) + if err != nil { + return ConfigResult{}, pkgerrors.Wrap(err, "Update Config DB Entry") + } + // Create Version Entry in DB for Config + cvs := ConfigVersionStore{ + rbName: rbName, + rbVersion: rbVersion, + profileName: profileName, + } + version, err := cvs.createConfigVersion(p, configPrev, "PUT") + if err != nil { + return ConfigResult{}, pkgerrors.Wrap(err, "Create Config Version DB Entry") + } + // Create Result structure + cfgRes := ConfigResult{ + DefinitionName: rbName, + DefinitionVersion: rbVersion, + ProfileName: profileName, + ConfigName: p.ConfigName, + TemplateName: p.TemplateName, + ConfigVersion: version, + } + return cfgRes, nil +} + +// Get config entry in the database +func (v *ConfigClient) Get(rbName, rbVersion, profileName, configName string) (Config, error) { + + // Acquire per profile Mutex + lock, _ := getProfileData(rbName + rbVersion + profileName) + lock.Lock() + defer lock.Unlock() + // Read Config DB + cs := ConfigStore{ + rbName: rbName, + rbVersion: rbVersion, + profileName: profileName, + configName: configName, + } + cfg, err := cs.getConfig() + if err != nil { + return Config{}, pkgerrors.Wrap(err, "Get Config DB Entry") + } + return cfg, nil +} + +// Delete the Config from database +func (v *ConfigClient) Delete(rbName, rbVersion, profileName, configName string) (ConfigResult, error) { + + // Check if Config exists + cs := ConfigStore{ + rbName: rbName, + rbVersion: rbVersion, + profileName: profileName, + configName: configName, + } + p, err := cs.getConfig() + if err != nil { + return ConfigResult{}, pkgerrors.Wrap(err, "Update Error - Config doesn't exist") + } + lock, profileChannel := getProfileData(rbName + rbVersion + profileName) + // Acquire per profile Mutex + lock.Lock() + defer lock.Unlock() + err = applyConfig(rbName, rbVersion, profileName, p, profileChannel, "DELETE") + if err != nil { + return ConfigResult{}, pkgerrors.Wrap(err, "Apply Config failed") + } + // Delete Config from DB + configPrev, err := cs.deleteConfig() + if err != nil { + return ConfigResult{}, pkgerrors.Wrap(err, "Delete Config DB Entry") + } + // Create Version Entry in DB for Config + cvs := ConfigVersionStore{ + rbName: rbName, + rbVersion: rbVersion, + profileName: profileName, + } + version, err := cvs.createConfigVersion(Config{}, configPrev, "DELETE") + if err != nil { + return ConfigResult{}, pkgerrors.Wrap(err, "Delete Config Version DB Entry") + } + // Create Result structure + cfgRes := ConfigResult{ + DefinitionName: rbName, + DefinitionVersion: rbVersion, + ProfileName: profileName, + ConfigName: configName, + TemplateName: configPrev.TemplateName, + ConfigVersion: version, + } + return cfgRes, nil +} + +// Rollback starts from current version and rollbacks to the version desired +func (v *ConfigClient) Rollback(rbName, rbVersion, profileName string, rback ConfigRollback) error { + + var reqVersion string + var err error + + if rback.AnyOf.ConfigTag != "" { + reqVersion, err = v.GetTagVersion(rbName, rbVersion, profileName, rback.AnyOf.ConfigTag) + if err != nil { + return pkgerrors.Wrap(err, "Rollback Invalid tag") + } + } else if rback.AnyOf.ConfigVersion != "" { + reqVersion = rback.AnyOf.ConfigVersion + } else { + return pkgerrors.Errorf("No valid Index for Rollback") + } + + index, err := strconv.Atoi(reqVersion) + if err != nil { + return pkgerrors.Wrap(err, "Rollback Invalid Index") + } + rollbackIndex := uint(index) + + lock, profileChannel := getProfileData(rbName + rbVersion + profileName) + // Acquire per profile Mutex + lock.Lock() + defer lock.Unlock() + + cvs := ConfigVersionStore{ + rbName: rbName, + rbVersion: rbVersion, + profileName: profileName, + } + currentVersion, err := cvs.getCurrentVersion() + if err != nil { + return pkgerrors.Wrap(err, "Rollback Get Current Config Version ") + } + + if rollbackIndex < 1 && rollbackIndex >= currentVersion { + return pkgerrors.Wrap(err, "Rollback Invalid Config Version") + } + + //Rollback all the intermettinent configurations + for i := currentVersion; i > rollbackIndex; i-- { + configNew, configPrev, action, err := cvs.getConfigVersion(i) + if err != nil { + return pkgerrors.Wrap(err, "Rollback Get Config Version") + } + cs := ConfigStore{ + rbName: rbName, + rbVersion: rbVersion, + profileName: profileName, + configName: configNew.ConfigName, + } + if action == "PUT" { + // PUT is proceeded by PUT or POST + err = applyConfig(rbName, rbVersion, profileName, configPrev, profileChannel, "PUT") + if err != nil { + return pkgerrors.Wrap(err, "Apply Config failed") + } + _, err = cs.updateConfig(configPrev) + if err != nil { + return pkgerrors.Wrap(err, "Update Config DB Entry") + } + } else if action == "POST" { + // POST is always preceeded by Config not existing + err = applyConfig(rbName, rbVersion, profileName, configNew, profileChannel, "DELETE") + if err != nil { + return pkgerrors.Wrap(err, "Delete Config failed") + } + _, err = cs.deleteConfig() + if err != nil { + return pkgerrors.Wrap(err, "Delete Config DB Entry") + } + } else if action == "DELETE" { + // DELETE is proceeded by PUT or POST + err = applyConfig(rbName, rbVersion, profileName, configPrev, profileChannel, "PUT") + if err != nil { + return pkgerrors.Wrap(err, "Delete Config failed") + } + _, err = cs.updateConfig(configPrev) + if err != nil { + return pkgerrors.Wrap(err, "Update Config DB Entry") + } + } + } + for i := currentVersion; i > rollbackIndex; i-- { + // Delete rolled back items + err = cvs.deleteConfigVersion() + if err != nil { + return pkgerrors.Wrap(err, "Delete Config Version ") + } + } + return nil +} + +// Tagit tags the current version with the tag provided +func (v *ConfigClient) Tagit(rbName, rbVersion, profileName string, tag ConfigTagit) error { + + lock, _ := getProfileData(rbName + rbVersion + profileName) + // Acquire per profile Mutex + lock.Lock() + defer lock.Unlock() + + cvs := ConfigVersionStore{ + rbName: rbName, + rbVersion: rbVersion, + profileName: profileName, + } + currentVersion, err := cvs.getCurrentVersion() + if err != nil { + return pkgerrors.Wrap(err, "Get Current Config Version ") + } + tagKey := constructKey(rbName, rbVersion, profileName, v.tagTag, tag.TagName) + + err = db.Etcd.Put(tagKey, strconv.Itoa(int(currentVersion))) + if err != nil { + return pkgerrors.Wrap(err, "TagIt store DB") + } + return nil +} + +// GetTagVersion returns the version associated with the tag +func (v *ConfigClient) GetTagVersion(rbName, rbVersion, profileName, tagName string) (string, error) { + + tagKey := constructKey(rbName, rbVersion, profileName, v.tagTag, tagName) + + value, err := db.Etcd.Get(tagKey) + if err != nil { + return "", pkgerrors.Wrap(err, "Config DB Entry Not found") + } + return string(value), nil +} + +// ApplyAllConfig starts from first configuration version and applies all versions in sequence +func (v *ConfigClient) ApplyAllConfig(rbName, rbVersion, profileName string) error { + + lock, profileChannel := getProfileData(rbName + rbVersion + profileName) + // Acquire per profile Mutex + lock.Lock() + defer lock.Unlock() + + cvs := ConfigVersionStore{ + rbName: rbName, + rbVersion: rbVersion, + profileName: profileName, + } + currentVersion, err := cvs.getCurrentVersion() + if err != nil { + return pkgerrors.Wrap(err, "Get Current Config Version ") + } + if currentVersion < 1 { + return pkgerrors.Wrap(err, "No Config Version to Apply") + } + //Apply all configurations + var i uint + for i = 1; i <= currentVersion; i++ { + configNew, _, action, err := cvs.getConfigVersion(i) + if err != nil { + return pkgerrors.Wrap(err, "Get Config Version") + } + err = applyConfig(rbName, rbVersion, profileName, configNew, profileChannel, action) + if err != nil { + return pkgerrors.Wrap(err, "Apply Config failed") + } + } + return nil +} diff --git a/src/k8splugin/internal/rb/config_backend.go b/src/k8splugin/internal/rb/config_backend.go new file mode 100644 index 00000000..b61fc493 --- /dev/null +++ b/src/k8splugin/internal/rb/config_backend.go @@ -0,0 +1,439 @@ +/* + * Copyright 2018 Intel Corporation, 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 + * + * http://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 rb + +import ( + "bytes" + "encoding/json" + "k8splugin/internal/db" + "k8splugin/internal/helm" + "log" + "strconv" + "strings" + + "io/ioutil" + "path/filepath" + "sync" + + "github.com/ghodss/yaml" + pkgerrors "github.com/pkg/errors" +) + +//ConfigStore contains the values that will be stored in the database +type configVersionDBContent struct { + ConfigNew Config `json:"config-new"` + ConfigPrev Config `json:"config-prev"` + Action string `json:"action"` // CRUD opration for this config +} + +//ConfigStore to Store the Config +type ConfigStore struct { + rbName string + rbVersion string + profileName string + configName string +} + +//ConfigVersionStore to Store the Versions of the Config +type ConfigVersionStore struct { + rbName string + rbVersion string + profileName string +} + +type configResourceList struct { + retmap map[string][]string + profile Profile + action string +} + +type profileDataManager struct { + profileLockMap map[string]*sync.Mutex + resourceChannel map[string](chan configResourceList) + sync.Mutex +} + +const ( + storeName = "config" + tagCounter = "counter" + tagVersion = "configversion" + tagConfig = "configdata" +) + +var profileData = profileDataManager{ + profileLockMap: map[string]*sync.Mutex{}, + resourceChannel: map[string]chan configResourceList{}, +} + +// Construct key for storing data +func constructKey(strs ...string) string { + + var sb strings.Builder + sb.WriteString("onapk8s") + sb.WriteString("/") + sb.WriteString(storeName) + sb.WriteString("/") + for _, str := range strs { + sb.WriteString(str) + sb.WriteString("/") + } + return sb.String() + +} + +// Create an entry for the config in the database +func (c ConfigStore) createConfig(p Config) error { + + cfgKey := constructKey(c.rbName, c.rbVersion, c.profileName, tagConfig, p.ConfigName) + _, err := db.Etcd.Get(cfgKey) + if err == nil { + return pkgerrors.Wrap(err, "Config DB Entry Already exists") + } + configValue, err := db.Serialize(p) + if err != nil { + return pkgerrors.Wrap(err, "Serialize Config Value") + } + err = db.Etcd.Put(cfgKey, configValue) + if err != nil { + return pkgerrors.Wrap(err, "Config DB Entry") + } + return nil +} + +// Update the config entry in the database. Updates with the new value +// Returns the previous value of the Config +func (c ConfigStore) updateConfig(p Config) (Config, error) { + + cfgKey := constructKey(c.rbName, c.rbVersion, c.profileName, tagConfig, p.ConfigName) + value, err := db.Etcd.Get(cfgKey) + configPrev := Config{} + if err == nil { + // If updating Config after rollback then previous config may not exist + err = db.DeSerialize(string(value), &configPrev) + if err != nil { + return Config{}, pkgerrors.Wrap(err, "DeSerialize Config Value") + } + } + configValue, err := db.Serialize(p) + if err != nil { + return Config{}, pkgerrors.Wrap(err, "Serialize Config Value") + } + err = db.Etcd.Put(cfgKey, configValue) + if err != nil { + return Config{}, pkgerrors.Wrap(err, "Config DB Entry") + } + return configPrev, nil +} + +// Read the config entry in the database +func (c ConfigStore) getConfig() (Config, error) { + cfgKey := constructKey(c.rbName, c.rbVersion, c.profileName, tagConfig, c.configName) + value, err := db.Etcd.Get(cfgKey) + if err != nil { + return Config{}, pkgerrors.Wrap(err, "Get Config DB Entry") + } + //value is a byte array + if value != nil { + cfg := Config{} + err = db.DeSerialize(string(value), &cfg) + if err != nil { + return Config{}, pkgerrors.Wrap(err, "Unmarshaling Config Value") + } + return cfg, nil + } + return Config{}, pkgerrors.Wrap(err, "Get Config DB Entry") +} + +// Delete the config entry in the database +func (c ConfigStore) deleteConfig() (Config, error) { + + cfgKey := constructKey(c.rbName, c.rbVersion, c.profileName, tagConfig, c.configName) + value, err := db.Etcd.Get(cfgKey) + if err != nil { + return Config{}, pkgerrors.Wrap(err, "Config DB Entry Not found") + } + configPrev := Config{} + err = db.DeSerialize(string(value), &configPrev) + if err != nil { + return Config{}, pkgerrors.Wrap(err, "DeSerialize Config Value") + } + + err = db.Etcd.Delete(cfgKey) + if err != nil { + return Config{}, pkgerrors.Wrap(err, "Config DB Entry") + } + return configPrev, nil +} + +// Create a version for the configuration. If previous config provided that is also stored +func (c ConfigVersionStore) createConfigVersion(configNew, configPrev Config, action string) (uint, error) { + + version, err := c.incrementVersion() + + if err != nil { + return 0, pkgerrors.Wrap(err, "Get Next Version") + } + versionKey := constructKey(c.rbName, c.rbVersion, c.profileName, tagVersion, strconv.Itoa(int(version))) + + var cs configVersionDBContent + cs.Action = action + cs.ConfigNew = configNew + cs.ConfigPrev = configPrev + + configValue, err := db.Serialize(cs) + if err != nil { + return 0, pkgerrors.Wrap(err, "Serialize Config Value") + } + err = db.Etcd.Put(versionKey, configValue) + if err != nil { + return 0, pkgerrors.Wrap(err, "Create Config DB Entry") + } + return version, nil +} + +// Delete current version of the configuration. Configuration always deleted from top +func (c ConfigVersionStore) deleteConfigVersion() error { + + counter, err := c.getCurrentVersion() + + if err != nil { + return pkgerrors.Wrap(err, "Get Next Version") + } + versionKey := constructKey(c.rbName, c.rbVersion, c.profileName, tagVersion, strconv.Itoa(int(counter))) + + err = db.Etcd.Delete(versionKey) + if err != nil { + return pkgerrors.Wrap(err, "Delete Config DB Entry") + } + err = c.decrementVersion() + if err != nil { + return pkgerrors.Wrap(err, "Decrement Version") + } + return nil +} + +// Read the specified version of the configuration and return its prev and current value. +// Also returns the action for the config version +func (c ConfigVersionStore) getConfigVersion(version uint) (Config, Config, string, error) { + + versionKey := constructKey(c.rbName, c.rbVersion, c.profileName, tagVersion, strconv.Itoa(int(version))) + configBytes, err := db.Etcd.Get(versionKey) + if err != nil { + return Config{}, Config{}, "", pkgerrors.Wrap(err, "Get Config Version ") + } + + if configBytes != nil { + pr := configVersionDBContent{} + err = db.DeSerialize(string(configBytes), &pr) + if err != nil { + return Config{}, Config{}, "", pkgerrors.Wrap(err, "DeSerialize Config Version") + } + return pr.ConfigNew, pr.ConfigPrev, pr.Action, nil + } + return Config{}, Config{}, "", pkgerrors.Wrap(err, "Invalid data ") +} + +// Get the counter for the version +func (c ConfigVersionStore) getCurrentVersion() (uint, error) { + + cfgKey := constructKey(c.rbName, c.rbVersion, c.profileName, tagCounter) + + value, err := db.Etcd.Get(cfgKey) + if err != nil { + if strings.Contains(err.Error(), "Key doesn't exist") == true { + // Counter not started yet, 0 is invalid value + return 0, nil + } else { + return 0, pkgerrors.Wrap(err, "Get Current Version") + } + } + + index, err := strconv.Atoi(string(value)) + if err != nil { + return 0, pkgerrors.Wrap(err, "Invalid counter") + } + return uint(index), nil +} + +// Update the counter for the version +func (c ConfigVersionStore) updateVersion(counter uint) error { + + cfgKey := constructKey(c.rbName, c.rbVersion, c.profileName, tagCounter) + err := db.Etcd.Put(cfgKey, strconv.Itoa(int(counter))) + if err != nil { + return pkgerrors.Wrap(err, "Counter DB Entry") + } + return nil +} + +// Increment the version counter +func (c ConfigVersionStore) incrementVersion() (uint, error) { + + counter, err := c.getCurrentVersion() + if err != nil { + return 0, pkgerrors.Wrap(err, "Get Next Counter Value") + } + //This is done while Profile lock is taken + counter++ + err = c.updateVersion(counter) + if err != nil { + return 0, pkgerrors.Wrap(err, "Store Next Counter Value") + } + + return counter, nil +} + +// Decrement the version counter +func (c ConfigVersionStore) decrementVersion() error { + + counter, err := c.getCurrentVersion() + if err != nil { + return pkgerrors.Wrap(err, "Get Next Counter Value") + } + //This is done while Profile lock is taken + counter-- + err = c.updateVersion(counter) + if err != nil { + return pkgerrors.Wrap(err, "Store Next Counter Value") + } + + return nil +} + +// Apply Config +func applyConfig(rbName, rbVersion, profileName string, p Config, pChannel chan configResourceList, action string) error { + + // Get Template and Resolve the template with values + crl, err := resolve(rbName, rbVersion, profileName, p) + if err != nil { + return pkgerrors.Wrap(err, "Resolve Config") + } + crl.action = action + // Send the configResourceList to the channel. Using select for non-blocking channel + select { + case pChannel <- crl: + log.Printf("Message Sent to goroutine %v", crl.profile) + default: + } + + return nil +} + +// Per Profile Go routine to apply the configuration to Cloud Region +func scheduleResources(c chan configResourceList) { + // Keep thread running + for { + data := <-c + //TODO: ADD Check to see if Application running + switch { + case data.action == "POST": + log.Printf("[scheduleResources]: POST %v %v", data.profile, data.retmap) + //TODO: Needs to add code to call Kubectl create + case data.action == "PUT": + log.Printf("[scheduleResources]: PUT %v %v", data.profile, data.retmap) + //TODO: Needs to add code to call Kubectl apply + case data.action == "DELETE": + log.Printf("[scheduleResources]: DELETE %v %v", data.profile, data.retmap) + //TODO: Needs to add code to call Kubectl delete + + } + } +} + +//Resolve returns the path where the helm chart merged with +//configuration overrides resides. +var resolve = func(rbName, rbVersion, profileName string, p Config) (configResourceList, error) { + + var retMap map[string][]string + + profile, err := NewProfileClient().Get(rbName, rbVersion, profileName) + if err != nil { + return configResourceList{}, pkgerrors.Wrap(err, "Reading Profile Data") + } + + t, err := NewConfigTemplateClient().Get(rbName, rbVersion, p.TemplateName) + if err != nil { + return configResourceList{}, pkgerrors.Wrap(err, "Getting Template") + } + if t.ChartName == "" { + return configResourceList{}, pkgerrors.New("Invalid template no Chart.yaml file found") + } + + def, err := NewConfigTemplateClient().Download(rbName, rbVersion, p.TemplateName) + if err != nil { + return configResourceList{}, pkgerrors.Wrap(err, "Downloading Template") + } + + //Create a temp file in the system temp folder for values input + b, err := json.Marshal(p.Values) + if err != nil { + return configResourceList{}, pkgerrors.Wrap(err, "Error Marshalling config data") + } + data, err := yaml.JSONToYAML(b) + if err != nil { + return configResourceList{}, pkgerrors.Wrap(err, "JSON to YAML") + } + + outputfile, err := ioutil.TempFile("", "helm-config-values-") + if err != nil { + return configResourceList{}, pkgerrors.Wrap(err, "Got error creating temp file") + } + _, err = outputfile.Write([]byte(data)) + if err != nil { + return configResourceList{}, pkgerrors.Wrap(err, "Got error writting temp file") + } + defer outputfile.Close() + + chartBasePath, err := ExtractTarBall(bytes.NewBuffer(def)) + if err != nil { + return configResourceList{}, pkgerrors.Wrap(err, "Extracting Template") + } + + helmClient := helm.NewTemplateClient(profile.KubernetesVersion, + profile.Namespace, + profile.ReleaseName) + + chartPath := filepath.Join(chartBasePath, t.ChartName) + retMap, err = helmClient.GenerateKubernetesArtifacts(chartPath, + []string{outputfile.Name()}, + nil) + if err != nil { + return configResourceList{}, pkgerrors.Wrap(err, "Generate final k8s yaml") + } + crl := configResourceList{ + retmap: retMap, + profile: profile, + } + + return crl, nil +} + +// Get the Mutex for the Profile +func getProfileData(key string) (*sync.Mutex, chan configResourceList) { + profileData.Lock() + defer profileData.Unlock() + _, ok := profileData.profileLockMap[key] + if !ok { + profileData.profileLockMap[key] = &sync.Mutex{} + } + _, ok = profileData.resourceChannel[key] + if !ok { + profileData.resourceChannel[key] = make(chan configResourceList) + go scheduleResources(profileData.resourceChannel[key]) + } + return profileData.profileLockMap[key], profileData.resourceChannel[key] +} diff --git a/src/k8splugin/internal/rb/config_template.go b/src/k8splugin/internal/rb/config_template.go new file mode 100644 index 00000000..cdb1b907 --- /dev/null +++ b/src/k8splugin/internal/rb/config_template.go @@ -0,0 +1,257 @@ +/* + * Copyright 2018 Intel Corporation, 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 + * + * http://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 rb + +import ( + "bytes" + "encoding/json" + "io/ioutil" + "k8splugin/internal/db" + "os" + "path/filepath" + + "encoding/base64" + + pkgerrors "github.com/pkg/errors" + "log" +) + +// ConfigTemplate contains the parameters needed for ConfigTemplates +type ConfigTemplate struct { + TemplateName string `json:"template-name"` + Description string `json:"description"` + ChartName string +} + +// ConfigTemplateManager is an interface exposes the resource bundle ConfigTemplate functionality +type ConfigTemplateManager interface { + Create(rbName, rbVersion string, p ConfigTemplate) error + Get(rbName, rbVersion, templateName string) (ConfigTemplate, error) + Delete(rbName, rbVersion, templateName string) error + Upload(rbName, rbVersion, templateName string, inp []byte) error +} + +// ConfigTemplateKey is key struct +type ConfigTemplateKey struct { + RBName string `json:"rb-name"` + RBVersion string `json:"rb-version"` + TemplateName string `json:"template-name"` +} + +// We will use json marshalling to convert to string to +// preserve the underlying structure. +func (dk ConfigTemplateKey) String() string { + out, err := json.Marshal(dk) + if err != nil { + return "" + } + + return string(out) +} + +// ConfigTemplateClient implements the ConfigTemplateManager +// It will also be used to maintain some localized state +type ConfigTemplateClient struct { + storeName string + tagMeta string + tagContent string +} + +// NewConfigTemplateClient returns an instance of the ConfigTemplateClient +// which implements the ConfigTemplateManager +func NewConfigTemplateClient() *ConfigTemplateClient { + return &ConfigTemplateClient{ + storeName: "rbdef", + tagMeta: "metadata", + tagContent: "content", + } +} + +// Create an entry for the resource bundle ConfigTemplate in the database +func (v *ConfigTemplateClient) Create(rbName, rbVersion string, p ConfigTemplate) error { + + log.Printf("[ConfigiTemplate]: create %s", rbName) + // Name is required + if p.TemplateName == "" { + return pkgerrors.New("Name is required for Resource Bundle ConfigTemplate") + } + + //Check if ConfigTemplate already exists + _, err := v.Get(rbName, rbVersion, p.TemplateName) + if err == nil { + return pkgerrors.New(" ConfigTemplate already exists for this Definition") + } + + //Check if provided resource bundle information is valid + _, err = NewDefinitionClient().Get(rbName, rbVersion) + if err != nil { + return pkgerrors.Errorf("Invalid Resource Bundle ID provided: %s", err.Error()) + } + + key := ConfigTemplateKey{ + RBName: rbName, + RBVersion: rbVersion, + TemplateName: p.TemplateName, + } + + err = db.DBconn.Create(v.storeName, key, v.tagMeta, p) + if err != nil { + return pkgerrors.Wrap(err, "Creating ConfigTemplate DB Entry") + } + + return nil +} + +// Get returns the Resource Bundle ConfigTemplate for corresponding ID +func (v *ConfigTemplateClient) Get(rbName, rbVersion, templateName string) (ConfigTemplate, error) { + key := ConfigTemplateKey{ + RBName: rbName, + RBVersion: rbVersion, + TemplateName: templateName, + } + value, err := db.DBconn.Read(v.storeName, key, v.tagMeta) + if err != nil { + return ConfigTemplate{}, pkgerrors.Wrap(err, "Get ConfigTemplate") + } + + //value is a byte array + if value != nil { + template := ConfigTemplate{} + err = db.DBconn.Unmarshal(value, &template) + if err != nil { + return ConfigTemplate{}, pkgerrors.Wrap(err, "Unmarshaling ConfigTemplate Value") + } + return template, nil + } + + return ConfigTemplate{}, pkgerrors.New("Error getting ConfigTemplate") +} + +// Delete the Resource Bundle ConfigTemplate from database +func (v *ConfigTemplateClient) Delete(rbName, rbVersion, templateName string) error { + key := ConfigTemplateKey{ + RBName: rbName, + RBVersion: rbVersion, + TemplateName: templateName, + } + err := db.DBconn.Delete(v.storeName, key, v.tagMeta) + if err != nil { + return pkgerrors.Wrap(err, "Delete ConfigTemplate") + } + + err = db.DBconn.Delete(v.storeName, key, v.tagContent) + if err != nil { + return pkgerrors.Wrap(err, "Delete ConfigTemplate Content") + } + + return nil +} + +// Upload the contents of resource bundle into database +func (v *ConfigTemplateClient) Upload(rbName, rbVersion, templateName string, inp []byte) error { + + log.Printf("[ConfigTemplate]: Upload %s", templateName) + key := ConfigTemplateKey{ + RBName: rbName, + RBVersion: rbVersion, + TemplateName: templateName, + } + //ignore the returned data here. + t, err := v.Get(rbName, rbVersion, templateName) + if err != nil { + return pkgerrors.Errorf("Invalid ConfigTemplate Name provided %s", err.Error()) + } + + err = isTarGz(bytes.NewBuffer(inp)) + if err != nil { + return pkgerrors.Errorf("Error in file format %s", err.Error()) + } + + chartBasePath, err := ExtractTarBall(bytes.NewBuffer(inp)) + if err != nil { + return pkgerrors.Wrap(err, "Extracting Template") + } + + finfo, err := ioutil.ReadDir(chartBasePath) + if err != nil { + return pkgerrors.Wrap(err, "Detecting chart name") + } + + //Store the first directory with Chart.yaml found as the chart name + for _, f := range finfo { + if f.IsDir() { + //Check if Chart.yaml exists + if _, err = os.Stat(filepath.Join(chartBasePath, f.Name(), "Chart.yaml")); err == nil { + t.ChartName = f.Name() + break + } + } + } + if t.ChartName == "" { + return pkgerrors.New("Invalid template no Chart.yaml file found") + } + + err = db.DBconn.Create(v.storeName, key, v.tagMeta, t) + if err != nil { + return pkgerrors.Wrap(err, "Creating ConfigTemplate DB Entry") + } + + //Encode given byte stream to text for storage + encodedStr := base64.StdEncoding.EncodeToString(inp) + err = db.DBconn.Create(v.storeName, key, v.tagContent, encodedStr) + if err != nil { + return pkgerrors.Errorf("Error uploading data to db %s", err.Error()) + } + + return nil +} + +// Download the contents of the ConfigTemplate from DB +// Returns a byte array of the contents +func (v *ConfigTemplateClient) Download(rbName, rbVersion, templateName string) ([]byte, error) { + + log.Printf("[ConfigTemplate]: Download %s", templateName) + //ignore the returned data here + //Check if rb is valid + _, err := v.Get(rbName, rbVersion, templateName) + if err != nil { + return nil, pkgerrors.Errorf("Invalid ConfigTemplate Name provided: %s", err.Error()) + } + + key := ConfigTemplateKey{ + RBName: rbName, + RBVersion: rbVersion, + TemplateName: templateName, + } + value, err := db.DBconn.Read(v.storeName, key, v.tagContent) + if err != nil { + return nil, pkgerrors.Wrap(err, "Get Resource ConfigTemplate content") + } + + if value != nil { + //Decode the string from base64 + out, err := base64.StdEncoding.DecodeString(string(value)) + if err != nil { + return nil, pkgerrors.Wrap(err, "Decode base64 string") + } + + if out != nil && len(out) != 0 { + return out, nil + } + } + return nil, pkgerrors.New("Error downloading ConfigTemplate content") +} diff --git a/src/k8splugin/internal/rb/config_test.go b/src/k8splugin/internal/rb/config_test.go new file mode 100644 index 00000000..9bf97a51 --- /dev/null +++ b/src/k8splugin/internal/rb/config_test.go @@ -0,0 +1,259 @@ +/* + * Copyright 2018 Intel Corporation, 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 + * + * http://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 rb + +import ( + "k8splugin/internal/db" + "reflect" + "strings" + "testing" + // pkgerrors "github.com/pkg/errors" +) + +func TestCreateConfig(t *testing.T) { + testCases := []struct { + label string + rbName string + rbVersion string + profileName string + inp Config + expectedError string + mockdb *db.MockEtcdClient + expected ConfigResult + }{ + { + label: "Create Config", + rbName: "testdef1", + rbVersion: "v1", + profileName: "testprofile1", + inp: Config{ + ConfigName: "testconfig1", + TemplateName: "testtemplate1", + Values: map[string]interface{}{ + "values": "{\"namespace\": \"kafka\", \"topic\": {\"name\":\"orders\", \"cluster\":\"my-cluster\", \"partitions\": 10,\"replicas\": 2, }}"}, + }, + expected: ConfigResult{ + DefinitionName: "testdef1", + DefinitionVersion: "v1", + ProfileName: "testprofile1", + ConfigName: "testconfig1", + TemplateName: "testtemplate1", + ConfigVersion: 1, + }, + expectedError: "", + mockdb: &db.MockEtcdClient{ + Items: nil, + Err: nil, + }, + }, + } + + for _, testCase := range testCases { + t.Run(testCase.label, func(t *testing.T) { + db.Etcd = testCase.mockdb + resolve = func(rbName, rbVersion, profileName string, p Config) (configResourceList, error) { + return configResourceList{}, nil + } + impl := NewConfigClient() + got, err := impl.Create(testCase.rbName, testCase.rbVersion, testCase.profileName, testCase.inp) + if err != nil { + if testCase.expectedError == "" { + t.Fatalf("Create returned an unexpected error %s", err) + } + if strings.Contains(err.Error(), testCase.expectedError) == false { + t.Fatalf("Create returned an unexpected error %s", err) + } + } else { + if reflect.DeepEqual(testCase.expected, got) == false { + t.Errorf("Create Resource Bundle returned unexpected body: got %v;"+ + " expected %v", got, testCase.expected) + } + } + }) + } +} + +func TestRollbackConfig(t *testing.T) { + testCases := []struct { + label string + rbName string + rbVersion string + profileName string + inp Config + inpUpdate1 Config + inpUpdate2 Config + expectedError string + mockdb *db.MockEtcdClient + expected1 ConfigResult + expected2 ConfigResult + expected3 ConfigResult + expected4 ConfigResult + rollbackConfig ConfigRollback + }{ + { + label: "Rollback Config", + rbName: "testdef1", + rbVersion: "v1", + profileName: "testprofile1", + inp: Config{ + ConfigName: "testconfig1", + TemplateName: "testtemplate1", + Values: map[string]interface{}{ + "values": "{\"namespace\": \"kafka\", \"topic\": {\"name\":\"orders\", \"cluster\":\"my-cluster\", \"partitions\": 10,\"replicas\": 2, }}"}, + }, + inpUpdate1: Config{ + ConfigName: "testconfig1", + TemplateName: "testtemplate1", + Values: map[string]interface{}{ + "values": "{\"namespace\": \"kafka\", \"topic\": {\"name\":\"orders\", \"cluster\":\"my-cluster\", \"partitions\": 20,\"replicas\": 2, }}"}, + }, + inpUpdate2: Config{ + ConfigName: "testconfig1", + TemplateName: "testtemplate1", + Values: map[string]interface{}{ + "values": "{\"namespace\": \"kafka\", \"topic\": {\"name\":\"orders\", \"cluster\":\"my-cluster\", \"partitions\": 30,\"replicas\": 2, }}"}, + }, + expected1: ConfigResult{ + DefinitionName: "testdef1", + DefinitionVersion: "v1", + ProfileName: "testprofile1", + ConfigName: "testconfig1", + TemplateName: "testtemplate1", + ConfigVersion: 1, + }, + expected2: ConfigResult{ + DefinitionName: "testdef1", + DefinitionVersion: "v1", + ProfileName: "testprofile1", + ConfigName: "testconfig1", + TemplateName: "testtemplate1", + ConfigVersion: 2, + }, + expected3: ConfigResult{ + DefinitionName: "testdef1", + DefinitionVersion: "v1", + ProfileName: "testprofile1", + ConfigName: "testconfig1", + TemplateName: "testtemplate1", + ConfigVersion: 3, + }, + expected4: ConfigResult{ + DefinitionName: "testdef1", + DefinitionVersion: "v1", + ProfileName: "testprofile1", + ConfigName: "testconfig1", + TemplateName: "testtemplate1", + ConfigVersion: 4, + }, + expectedError: "", + mockdb: &db.MockEtcdClient{ + Items: nil, + Err: nil, + }, + }, + } + + for _, testCase := range testCases { + t.Run(testCase.label, func(t *testing.T) { + db.Etcd = testCase.mockdb + resolve = func(rbName, rbVersion, profileName string, p Config) (configResourceList, error) { + return configResourceList{}, nil + } + impl := NewConfigClient() + got, err := impl.Create(testCase.rbName, testCase.rbVersion, testCase.profileName, testCase.inp) + if err != nil { + if testCase.expectedError == "" { + t.Fatalf("Create returned an unexpected error %s", err) + } + if strings.Contains(err.Error(), testCase.expectedError) == false { + t.Fatalf("Create returned an unexpected error %s", err) + } + } else { + if reflect.DeepEqual(testCase.expected1, got) == false { + t.Errorf("Create Resource Bundle returned unexpected body: got %v;"+ + " expected %v", got, testCase.expected1) + } + } + got, err = impl.Update(testCase.rbName, testCase.rbVersion, testCase.profileName, testCase.inp.ConfigName, testCase.inpUpdate1) + if err != nil { + if testCase.expectedError == "" { + t.Fatalf("Create returned an unexpected error %s", err) + } + if strings.Contains(err.Error(), testCase.expectedError) == false { + t.Fatalf("Create returned an unexpected error %s", err) + } + } else { + if reflect.DeepEqual(testCase.expected2, got) == false { + t.Errorf("Create Resource Bundle returned unexpected body: got %v;"+ + " expected %v", got, testCase.expected2) + } + } + got, err = impl.Update(testCase.rbName, testCase.rbVersion, testCase.profileName, testCase.inp.ConfigName, testCase.inpUpdate2) + if err != nil { + if testCase.expectedError == "" { + t.Fatalf("Create returned an unexpected error %s", err) + } + if strings.Contains(err.Error(), testCase.expectedError) == false { + t.Fatalf("Create returned an unexpected error %s", err) + } + } else { + if reflect.DeepEqual(testCase.expected3, got) == false { + t.Errorf("Create Resource Bundle returned unexpected body: got %v;"+ + " expected %v", got, testCase.expected3) + } + } + got, err = impl.Delete(testCase.rbName, testCase.rbVersion, testCase.profileName, testCase.inp.ConfigName) + if err != nil { + if testCase.expectedError == "" { + t.Fatalf("Create returned an unexpected error %s", err) + } + if strings.Contains(err.Error(), testCase.expectedError) == false { + t.Fatalf("Create returned an unexpected error %s", err) + } + } else { + if reflect.DeepEqual(testCase.expected4, got) == false { + t.Errorf("Create Resource Bundle returned unexpected body: got %v;"+ + " expected %v", got, testCase.expected4) + } + } + testCase.rollbackConfig.AnyOf.ConfigVersion = "2" + err = impl.Rollback(testCase.rbName, testCase.rbVersion, testCase.profileName, testCase.rollbackConfig) + if err != nil { + if testCase.expectedError == "" { + t.Fatalf("Create returned an unexpected error %s", err) + } + if strings.Contains(err.Error(), testCase.expectedError) == false { + t.Fatalf("Create returned an unexpected error %s", err) + } + } + rollbackConfig, err := impl.Get(testCase.rbName, testCase.rbVersion, testCase.profileName, testCase.inp.ConfigName) + if err != nil { + if testCase.expectedError == "" { + t.Fatalf("Create returned an unexpected error %s", err) + } + if strings.Contains(err.Error(), testCase.expectedError) == false { + t.Fatalf("Create returned an unexpected error %s", err) + } + } else { + if reflect.DeepEqual(testCase.inpUpdate1, rollbackConfig) == false { + t.Errorf("Rollback config failed: got %v;"+ + " expected %v", rollbackConfig, testCase.inpUpdate1) + } + } + }) + } +} |