This rather large update adds the following
- A golang build container used for building golang executables and/or
containers.
- envoyd, a daemon process that creates and updates the envoy config
file based on consul's KV store and forces envoy to reload the config
as it changes.
- Dockerfile(s) and compose files that integrate envoy into the NBI call
chain to load-balance device-to-core assignment.
- Several developer tools that help build and replace specific
containers in a running cluster. This allows the build process to be
separated from the run-time as it will be in production.
- NOTES: A command line needs to be added to envoyd because now the
values are declared at the start of the file. This will be submitted
in a subsequent commit along with a change toward a more object
oriented implementation.
Addressed reviewer comments.
Addressed even more reviewr comments.
Change-Id: Ia2ec825d48d475398e501f396452fb0306673432
diff --git a/envoy/go/envoyd/envoyd.go b/envoy/go/envoyd/envoyd.go
new file mode 100644
index 0000000..6766539
--- /dev/null
+++ b/envoy/go/envoyd/envoyd.go
@@ -0,0 +1,335 @@
+package main // import "ciena.com/envoyd"
+
+import (
+ "os"
+ "os/exec"
+ "fmt"
+ "log"
+ "strconv"
+ "time"
+ "net"
+ "io/ioutil"
+ "text/template"
+ "encoding/json"
+ consulapi "github.com/hashicorp/consul/api"
+)
+
+// DATA STRUCTURES
+
+type ConfigVars struct {
+ VolthaVip string
+ VolthaRR []string
+}
+
+type VolthaClusterEntry struct {
+ Prefix string
+ Id string
+ Host string
+}
+
+//Client provides an interface for getting data out of Consul
+type Client interface {
+// Get a Service from consulapi
+ Service(string, string) ([]string, error)
+// Register a service with local agent
+ Register(string, int) error
+// Deregister a service with local agent
+ DeRegister(string) error
+}
+
+type client struct {
+ consulapi *consulapi.Client
+}
+
+// This struct is not used yet
+// TODO: Update the daemon to use this structure to for a
+// more object oriented implementation
+type EnvoyControl struct {
+ retrys int
+ waitTime int
+ cv ConfigVars
+ vc []VolthaClusterEntry
+ meta * consulapi.QueryMeta
+ kvp * consulapi.KVPair
+ ipAddrs map[string][]string
+}
+
+// CONSTANTS
+var assignmentKey string = "service/voltha/data/core/assignment"
+var vcoreHostIpName string = "host"
+var vcoreIdName string = "id"
+var restartEpoch int = 0
+var volthaPort string = "50556" // This will be passed inas an option.
+var consulPort string = "8500" // This will be passed in as an option.
+
+//NewConsul returns a Client interface for given consulapi address
+func NewConsulClient(addr string) (*client, error) {
+ config := consulapi.DefaultConfig()
+ config.Address = addr
+ c, err := consulapi.NewClient(config)
+ if err != nil {
+ return nil, err
+ }
+ return &client{consulapi: c}, nil
+}
+
+// Register a service with consulapi local agent
+func (c *client) Register(name string, port int) error {
+ reg := &consulapi.AgentServiceRegistration{
+ ID: name,
+ Name: name,
+ Port: port,
+ }
+ return c.consulapi.Agent().ServiceRegister(reg)
+}
+
+// DeRegister a service with consulapi local agent
+func (c *client) DeRegister(id string) error {
+ return c.consulapi.Agent().ServiceDeregister(id)
+}
+
+// Service return a service
+func (c *client) Service(service, tag string) ([]*consulapi.ServiceEntry, *consulapi.QueryMeta, error) {
+ passingOnly := true
+ addrs, meta, err := c.consulapi.Health().Service(service, tag, passingOnly, nil)
+ if len(addrs) == 0 && err == nil {
+ return nil, nil, fmt.Errorf("service ( %s ) was not found", service)
+ }
+ if err != nil {
+ return nil, nil, err
+ }
+ return addrs, meta, nil
+}
+
+// Starts envoy with the current restartEpoch
+func startEnvoy(cfg_file string) {
+ cmd := exec.Command("/usr/local/bin/envoy", "--restart-epoch", strconv.Itoa(restartEpoch),
+ "--config-path", cfg_file)
+
+ curEpoch := restartEpoch
+ restartEpoch += 1
+ if err := cmd.Start(); err != nil {
+ log.Fatal(err)
+ panic(err)
+ }
+ log.Printf("Waiting on envoy %d to exit", curEpoch)
+ if err := cmd.Wait(); err != nil {
+ log.Fatal(err, "Unexpected exit code")
+ }
+ log.Printf("Envoy %d exited", curEpoch)
+
+}
+
+// This function will use the provided templete file to generate
+// the targetfile substituting
+func updateEnvoyConfig(templateFile string, targetFile string, cv ConfigVars) {
+ var firstRun bool = true
+ f := func() (bool) {
+ var rtrn bool = firstRun
+ firstRun = false
+ return rtrn
+ }
+ var funcs = template.FuncMap{"isFirst": f}
+ // Slurp up the template file.
+ tplt, err := ioutil.ReadFile(templateFile)
+ if err != nil {
+ panic("ERROR reading the template file, aborting")
+ }
+ //fmt.Println(string(tplt))
+ configTemplate, err := template.New("config").Funcs(funcs).Parse(string(tplt));
+ if err != nil {
+ panic(err)
+ }
+ outFile,err := os.Create(targetFile)
+ if err != nil {
+ panic(err)
+ }
+ if err := configTemplate.Execute(outFile, cv); err != nil {
+ panic(err)
+ }
+ //cfgFile, err := ioutil.ReadFile(targetFile)
+ //if err != nil {
+ // panic("ERROR reading the config file, aborting")
+ //}
+ //fmt.Println(string(cfgFile))
+}
+
+func getServiceAddr(serviceName string, retrys int, waitTime int) (addrs []string, err error) {
+ for i := 0; i < retrys; i++ {
+ addrs,err = net.LookupHost(serviceName)
+ if err != nil {
+ log.Printf("%s name resolution failed %d time(s) retrying...\n", serviceName, i+1)
+ } else {
+ //fmt.Printf("%s address = %s\n",serviceName, addrs[0])
+ break
+ }
+ time.Sleep(time.Duration(waitTime) * time.Second)
+ }
+ if err != nil {
+ log.Printf("%s name resolution failed %d times gving up\n", serviceName, retrys)
+ }
+ return
+}
+
+func parseAssignment(jsonString []byte) (vCluster []VolthaClusterEntry, err error) {
+ var vc VolthaClusterEntry
+ var f interface{}
+
+ log.Printf("Parsing %s\n", string(jsonString))
+ err = json.Unmarshal(jsonString, &f)
+ if err != nil {
+ log.Fatal("Unable to parse json record %s", jsonString)
+ panic(err)
+ } else {
+ m := f.(map[string]interface{})
+ for k, v := range m {
+ vc.Prefix = k
+ //log.Printf("Processing key %s\n", k)
+ switch vv := v.(type) {
+ case map[string]interface{}:
+ for i, u := range vv {
+ //log.Printf("Processing key %s\n", i)
+ switch uu := u.(type) {
+ case string:
+ if i == vcoreHostIpName {
+ vc.Host = uu
+ } else if i == vcoreIdName {
+ vc.Id = uu
+ } else {
+ log.Printf("WARNING: unexpected descriptor,%s\n", i)
+ }
+ default:
+ log.Printf("WARNING: unexpected type, ")
+ log.Println(i, u)
+ }
+ }
+ default:
+ log.Printf("WARNING: unexpected type, ")
+ log.Println(k, v)
+ }
+ vCluster = append(vCluster, vc)
+ }
+ }
+ log.Println("Parsing complete")
+ return
+}
+
+func runEnvoy(meta * consulapi.QueryMeta, kvp * consulapi.KVPair, values * map[string]interface{}, cv ConfigVars,
+ templatePath string, configPath string) {
+ var err error
+ var vCluster []VolthaClusterEntry
+
+ // Extract all values from the KV record
+ vCluster, err = parseAssignment([]byte(kvp.Value))
+ if err == nil {
+ (*values)["volthaRR"] = []string{}
+ for i := range vCluster {
+ //log.Printf("Processing %s\n", vCluster[i].Host)
+ (*values)["volthaRR"] = append((*values)["volthaRR"].([]string), vCluster[i].Host)
+ cv.VolthaRR = append(cv.VolthaRR, vCluster[i].Host + ":" + volthaPort)
+ }
+ } else {
+ log.Fatal("Couldn't parse the KV record %s\n", string(kvp.Value))
+ panic(err)
+ }
+
+ // Now that we have the data loaded, update the envoy config and start envoy
+ updateEnvoyConfig(templatePath, configPath, cv)
+ go startEnvoy(configPath)
+ log.Printf("meta.LastIndex = %d\n", meta.LastIndex)
+}
+
+func runMonitorEnvoy(kv * consulapi.KV, values * map[string]interface{}, cv ConfigVars,
+ templatePath string, configPath string) {
+ var err error
+ var qo consulapi.QueryOptions
+
+
+ // Get the initial values of the assignment key which contains individual
+ // voltha core IP addresses. This may be empty until voltha populates it
+ // so it must be checked
+ kvp, meta, err := kv.Get(assignmentKey, nil)
+ for i := 0; i < 10; i++ {
+ if err != nil {
+ fmt.Println(err)
+ log.Printf("Unable to read assignment consul key, retry %d\n", i+1)
+ time.Sleep(time.Duration(2) * time.Second)
+ kvp, meta, err = kv.Get(assignmentKey, nil)
+ } else if kvp != nil && len(string(kvp.Value)) > 10 {
+ log.Printf("Starting Envoy")
+ runEnvoy(meta, kvp, values, cv, templatePath, configPath)
+ break
+ } else {
+ log.Printf("Voltha assignment key invalid, retry %d\n", i+1)
+ time.Sleep(time.Duration(2) * time.Second)
+ kvp, meta, err = kv.Get(assignmentKey, nil)
+ }
+ }
+
+ for {
+ qo.WaitIndex = meta.LastIndex
+ for {
+ if qo.WaitIndex != meta.LastIndex {
+ break
+ }
+ kvp, meta, err = kv.Get(assignmentKey, &qo)
+ if err != nil {
+ log.Fatal("Unable to read assignment consul key")
+ panic(err)
+ } else {
+ log.Println(string(kvp.Value))
+ log.Printf("meta.LastIndex = %d\n", meta.LastIndex)
+ }
+ }
+ // Fell through, the index has changed thus the key has changed
+
+ runEnvoy(meta, kvp, values, cv, templatePath, configPath)
+ }
+}
+
+func main() {
+
+ var err error
+ var addrs []string
+ var cv ConfigVars // Template variables.
+ var consul * consulapi.Client
+ var values map[string]interface{} // Key values map
+
+ values = make(map[string]interface{})
+
+ // Resolve consul's virtual ip address
+ addrs, err = getServiceAddr("consul", 10, 2)
+ if err == nil {
+ values["consulvip"] = addrs[0]
+ log.Printf("Consul's address = %s\n",addrs[0])
+ } else {
+ log.Fatal("Can't proceed without consul's vIP address")
+ panic(err)
+ }
+
+ // Resolve voltha's virtual ip address
+ addrs,err = getServiceAddr("vcore", 10, 2)
+ if err == nil {
+ log.Printf("Voltha address = %s\n",addrs[0])
+ // Config var for the template
+ cv.VolthaVip = addrs[0] + ":" + volthaPort
+ values["volthavip"] = addrs[0]
+ } else {
+ log.Fatal("Can't proceed without voltha's vIP address")
+ panic(err)
+ }
+
+ // Fire up a consul client and get the kv store
+ config := consulapi.DefaultConfig()
+ config.Address = values["consulvip"].(string) + ":" + consulPort
+ consul, err = consulapi.NewClient(config)
+ if err != nil {
+ log.Fatal("error creating consul client aborting")
+ panic(err)
+ }
+ kv := consul.KV()
+
+ // Start envoy and monitor changes to the KV store to reload
+ // consul's config. This never returns unless somethign crashes.
+ runMonitorEnvoy(kv, &values, cv, "/envoy/voltha-grpc-proxy.template.json", "/envoy/voltha-grpc-proxy.json")
+}