ip2etcd_freeflow.go

// Copyright (c) 2018 Gang Liao <gangliao@cs.umd.edu>. All Rights Reserved.

// 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 main

import (
	"context"
	"crypto/tls"
	"crypto/x509"
	"flag"
	"fmt"
	"io/ioutil"
	"os"
	"path/filepath"
	"strings"
	"time"

	"github.com/coreos/etcd/clientv3"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/clientcmd"
)

var (
	cert   *string
	key    *string
	caCert *string

	kubeconfig *string

	ipMap = make(map[string]string)
)

const (
	updateTime           = 1 * time.Second
	requestTimeout       = 5 * time.Second
	dialKeepAliveTime    = 10 * time.Second
	dialKeepAliveTimeout = 3 * time.Second
	keyDir               = "Microsoft/FreeFlow/"
)

func homeDir() string {
	if h := os.Getenv("HOME"); h != "" {
		return h
	}
	return os.Getenv("USERPROFILE") // windows
}

func main() {
	if home := homeDir(); home != "" {
		kubeconfig = flag.String("kubeconfig", filepath.Join(home, ".kube", "config"), "(optional) absolute path to the kubeconfig file")
	} else {
		kubeconfig = flag.String("kubeconfig", "", "absolute path to the kubeconfig file")
	}

	caCert = flag.String("cacert", "", "etcd cacert")
	cert = flag.String("cert", "", "etcd cert")
	key = flag.String("key", "", "etcd key")

	endpoints := strings.Split(*flag.String("endpoints", os.Getenv("ETCD_ENDPOINTS"), "etcd endpoints"), ",")
	namespace := flag.String("namespace", "default", "name space to query")

	fmt.Println("endpoints: ", endpoints)
	flag.Parse()

	// use the current context in kubeconfig
	config, err := clientcmd.BuildConfigFromFlags("", *kubeconfig)
	if err != nil {
		panic(err.Error())
	}

	// creates the clientset
	clientset, err := kubernetes.NewForConfig(config)
	if err != nil {
		panic(err.Error())
	}

	updateIP(namespace, clientset, endpoints)
}

func etcdClient(endpoints []string) *clientv3.Client {
	cfg := clientv3.Config{
		Endpoints:            endpoints,
		DialTimeout:          requestTimeout,
		DialKeepAliveTime:    dialKeepAliveTime,
		DialKeepAliveTimeout: dialKeepAliveTimeout,
	}

	tlsEnabled := false
	tlsConfig := &tls.Config{
		InsecureSkipVerify: false,
	}

	if *caCert != "" {
		certBytes, err := ioutil.ReadFile(*caCert)
		if err != nil {
			panic(err.Error())
		}

		caCertPool := x509.NewCertPool()
		ok := caCertPool.AppendCertsFromPEM(certBytes)

		if ok {
			tlsConfig.RootCAs = caCertPool
		}
		tlsEnabled = true
	}

	if *cert != "" && *key != "" {
		tlsCert, err := tls.LoadX509KeyPair(*cert, *key)
		if err != nil {
			panic(err.Error())
		}
		tlsConfig.Certificates = []tls.Certificate{tlsCert}
		tlsEnabled = true
	}

	if tlsEnabled {
		cfg.TLS = tlsConfig
	}

	clientetcd, err := clientv3.New(cfg)
	if err != nil {
		panic(err.Error())
	}

	return clientetcd
}

func updateIP(namespace *string, clientset *kubernetes.Clientset, endpoints []string) {
	clientetcd := etcdClient(endpoints)
	defer clientetcd.Close()

	for true {
		// fetch pod information
		pods, err := clientset.CoreV1().Pods(*namespace).List(metav1.ListOptions{})
		if err != nil {
			panic(err.Error())
		}

		// update ip into new map
		ipNewMap := make(map[string]string)
		for _, pod := range pods.Items {
			if hostOldIP, ok := ipMap[pod.Status.PodIP]; ok {
				if hostOldIP == pod.Status.HostIP {
					continue
				}
			}
			fmt.Println(pod.Status.PodIP + " <- " + pod.Status.HostIP)
			ipMap[pod.Status.PodIP] = pod.Status.HostIP
			ipNewMap[pod.Status.PodIP] = pod.Status.HostIP
		}

		// delete key-value ip map in etcd recursively
		if len(ipNewMap) != 0 {
			_, err := clientetcd.Delete(context.Background(), keyDir, clientv3.WithPrefix())
			if err != nil {
				panic(err.Error())
			}

			for k, v := range ipNewMap {
				_, err := clientetcd.Put(context.Background(), keyDir+k, v)
				if err != nil {
					panic(err.Error())
				}
			}
		}

		time.Sleep(updateTime)
	}

}

 go run etcd.go -cacert=/etc/kubernetes/ssl/ca.pem -cert=/etc/etcd/ssl/etcd.pem -key=/etc/etcd/ssl/etcd-key.pem

$ETCDCTL_API=3
$ etcdctl get --prefix Microsoft

Microsoft/FreeFlow/172.17.0.2
10.141.177.102
Microsoft/FreeFlow/172.17.0.3
10.141.8.29
Microsoft/FreeFlow/172.17.0.4
10.141.187.21