From 401aba14c5f5e55480afb491af2bf953cabc6ac2 Mon Sep 17 00:00:00 2001 From: Kiran Kamineni Date: Wed, 15 May 2019 16:47:39 -0700 Subject: Move config to app and connect to instance Move config instantiation to app and connect it to the instance to allow updates and so on. Issue-ID: MULTICLOUD-464 Change-Id: Ic994ef78a6e0d2db5e695e33b7b8a302c74c10da Signed-off-by: Kiran Kamineni --- src/k8splugin/internal/app/config_backend.go | 475 +++++++++++++++++++++++++++ 1 file changed, 475 insertions(+) create mode 100644 src/k8splugin/internal/app/config_backend.go (limited to 'src/k8splugin/internal/app/config_backend.go') diff --git a/src/k8splugin/internal/app/config_backend.go b/src/k8splugin/internal/app/config_backend.go new file mode 100644 index 00000000..763aed0d --- /dev/null +++ b/src/k8splugin/internal/app/config_backend.go @@ -0,0 +1,475 @@ +/* + * 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 app + +import ( + "bytes" + "encoding/json" + "io/ioutil" + "log" + "path/filepath" + "strconv" + "strings" + "sync" + "time" + + "k8splugin/internal/db" + "k8splugin/internal/helm" + "k8splugin/internal/rb" + + "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 { + resourceTemplates []helm.KubernetesResourceTemplate + createdResources []helm.KubernetesResource + profile rb.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 + ic := NewInstanceClient() + resp, err := ic.Find(data.profile.RBName, data.profile.RBVersion, data.profile.ProfileName) + if err != nil || len(resp) == 0 { + log.Println("Error finding a running instance. Retrying later...") + time.Sleep(time.Second * 10) + continue + } + switch { + case data.action == "POST": + log.Printf("[scheduleResources]: POST %v %v", data.profile, data.resourceTemplates) + for _, inst := range resp { + k8sClient := KubernetesClient{} + err = k8sClient.init(inst.CloudRegion) + if err != nil { + log.Printf("Getting CloudRegion Information: %s", err.Error()) + //Move onto the next cloud region + continue + } + data.createdResources, err = k8sClient.createResources(data.resourceTemplates, inst.Namespace) + if err != nil { + log.Printf("Error Creating resources: %s", err.Error()) + continue + } + } + //TODO: Needs to add code to call Kubectl create + case data.action == "PUT": + log.Printf("[scheduleResources]: PUT %v %v", data.profile, data.resourceTemplates) + //TODO: Needs to add code to call Kubectl apply + case data.action == "DELETE": + log.Printf("[scheduleResources]: DELETE %v %v", data.profile, data.resourceTemplates) + for _, inst := range resp { + k8sClient := KubernetesClient{} + err = k8sClient.init(inst.CloudRegion) + if err != nil { + log.Printf("Getting CloudRegion Information: %s", err.Error()) + //Move onto the next cloud region + continue + } + err = k8sClient.deleteResources(data.createdResources, inst.Namespace) + if err != nil { + log.Printf("Error Deleting resources: %s", err.Error()) + continue + } + } + } + } +} + +//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 resTemplates []helm.KubernetesResourceTemplate + + profile, err := rb.NewProfileClient().Get(rbName, rbVersion, profileName) + if err != nil { + return configResourceList{}, pkgerrors.Wrap(err, "Reading Profile Data") + } + + t, err := rb.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 := rb.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 := rb.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) + resTemplates, err = helmClient.GenerateKubernetesArtifacts(chartPath, + []string{outputfile.Name()}, + nil) + if err != nil { + return configResourceList{}, pkgerrors.Wrap(err, "Generate final k8s yaml") + } + crl := configResourceList{ + resourceTemplates: resTemplates, + 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] +} -- cgit 1.2.3-korg