envoy/go/envoyd/envoyd.go - voltha - Gitiles

 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")
 }
	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")
	}