<commit_before>package main

import (
	"encoding/json"
	"errors"
	"flag"
	"fmt"
	"log"
	"net/http"
	"strings"

	"github.com/pborman/uuid"
	"k8s.io/kubernetes/pkg/api"
	"k8s.io/kubernetes/pkg/client/restclient"
	"k8s.io/kubernetes/pkg/client/unversioned"
	"k8s.io/kubernetes/pkg/util/intstr"
)

var masterHost string

func init() {
	flag.StringVar(&masterHost, "master", "http://127.0.0.1:8080", "TODO: usage")
	flag.Parse()
}

type etcdClusterController struct {
	kclient *unversioned.Client
}

func (c *etcdClusterController) Run() {
	eventCh, errCh := monitorNewCluster()
	for {
		select {
		case event := <-eventCh:
			c.createCluster(event)
		case err := <-errCh:
			panic(err)
		}
	}
}

func (c *etcdClusterController) createCluster(event newCluster) {
	size := event.Size
	clusterName := event.Metadata["name"]

	initialCluster := []string{}
	for i := 0; i < size; i++ {
		initialCluster = append(initialCluster, fmt.Sprintf("%s-%04d=http://%s-%04d:2380", clusterName, i, clusterName, i))
	}

	for i := 0; i < size; i++ {
		etcdName := fmt.Sprintf("%s-%04d", clusterName, i)

		svc := makeEtcdService(etcdName, clusterName)
		_, err := c.kclient.Services("default").Create(svc)
		if err != nil {
			panic(err)
		}
		// TODO: add and expose client port
		pod := makeEtcdPod(etcdName, clusterName, initialCluster)
		_, err = c.kclient.Pods("default").Create(pod)
		if err != nil {
			panic(err)
		}
	}
}

type newCluster struct {
	Kind       string            `json:"kind"`
	ApiVersion string            `json:"apiVersion"`
	Metadata   map[string]string `json:"metadata"`
	Size       int               `json:"size"`
}

type Event struct {
	Type   string
	Object newCluster
}

func monitorNewCluster() (<-chan newCluster, <-chan error) {
	events := make(chan newCluster)
	errc := make(chan error, 1)
	go func() {
		resp, err := http.Get(masterHost + "/apis/coreos.com/v1/namespaces/default/etcdclusters?watch=true")
		if err != nil {
			errc <- err
			return
		}
		if resp.StatusCode != 200 {
			errc <- errors.New("Invalid status code: " + resp.Status)
			return
		}
		log.Println("start watching...")
		for {
			decoder := json.NewDecoder(resp.Body)
			var ev Event
			err = decoder.Decode(&ev)
			if err != nil {
				errc <- err
			}
			event := ev.Object
			log.Println("new cluster size:", event.Size, event.Metadata)
			events <- event
		}
	}()

	return events, errc
}

func main() {
	c := &etcdClusterController{
		kclient: mustCreateClient(masterHost),
	}
	log.Println("etcd cluster controller starts running...")
	c.Run()
}

func mustCreateClient(host string) *unversioned.Client {
	cfg := &restclient.Config{
		Host:  host,
		QPS:   100,
		Burst: 100,
	}
	c, err := unversioned.New(cfg)
	if err != nil {
		panic(err)
	}
	return c
}

func generateUUID() string {
	return uuid.New()
}

func makeEtcdService(etcdName, uuid string) *api.Service {
	labels := map[string]string{
		"etcd_node": etcdName,
		"etcd_uuid": uuid,
	}
	svc := &api.Service{
		ObjectMeta: api.ObjectMeta{
			Name:   etcdName,
			Labels: labels,
		},
		Spec: api.ServiceSpec{
			Ports: []api.ServicePort{{
				Name:       "server",
				Port:       2380,
				TargetPort: intstr.FromInt(2380),
				Protocol:   api.ProtocolTCP,
			}},
			Selector: labels,
		},
	}
	return svc
}

func makeEtcdPod(etcdName, uuid string, initialCluster []string) *api.Pod {
	pod := &api.Pod{
		ObjectMeta: api.ObjectMeta{
			Name: etcdName,
			Labels: map[string]string{
				"app":       "etcd",
				"etcd_node": etcdName,
				"etcd_uuid": uuid,
			},
		},
		Spec: api.PodSpec{
			Containers: []api.Container{
				{
					Command: []string{
						"/usr/local/bin/etcd",<commit_msg>fix: uuid -> cluster<commit_after>package main

import (
	"encoding/json"
	"errors"
	"flag"
	"fmt"
	"log"
	"net/http"
	"strings"

	"k8s.io/kubernetes/pkg/api"
	"k8s.io/kubernetes/pkg/client/restclient"
	"k8s.io/kubernetes/pkg/client/unversioned"
	"k8s.io/kubernetes/pkg/util/intstr"
)

var masterHost string

func init() {
	flag.StringVar(&masterHost, "master", "http://127.0.0.1:8080", "TODO: usage")
	flag.Parse()
}

type etcdClusterController struct {
	kclient *unversioned.Client
}

func (c *etcdClusterController) Run() {
	eventCh, errCh := monitorNewCluster()
	for {
		select {
		case event := <-eventCh:
			c.createCluster(event)
		case err := <-errCh:
			panic(err)
		}
	}
}

func (c *etcdClusterController) createCluster(event newCluster) {
	size := event.Size
	clusterName := event.Metadata["name"]

	initialCluster := []string{}
	for i := 0; i < size; i++ {
		initialCluster = append(initialCluster, fmt.Sprintf("%s-%04d=http://%s-%04d:2380", clusterName, i, clusterName, i))
	}

	for i := 0; i < size; i++ {
		etcdName := fmt.Sprintf("%s-%04d", clusterName, i)

		svc := makeEtcdService(etcdName, clusterName)
		_, err := c.kclient.Services("default").Create(svc)
		if err != nil {
			panic(err)
		}
		// TODO: add and expose client port
		pod := makeEtcdPod(etcdName, clusterName, initialCluster)
		_, err = c.kclient.Pods("default").Create(pod)
		if err != nil {
			panic(err)
		}
	}
}

type newCluster struct {
	Kind       string            `json:"kind"`
	ApiVersion string            `json:"apiVersion"`
	Metadata   map[string]string `json:"metadata"`
	Size       int               `json:"size"`
}

type Event struct {
	Type   string
	Object newCluster
}

func monitorNewCluster() (<-chan newCluster, <-chan error) {
	events := make(chan newCluster)
	errc := make(chan error, 1)
	go func() {
		resp, err := http.Get(masterHost + "/apis/coreos.com/v1/namespaces/default/etcdclusters?watch=true")
		if err != nil {
			errc <- err
			return
		}
		if resp.StatusCode != 200 {
			errc <- errors.New("Invalid status code: " + resp.Status)
			return
		}
		log.Println("start watching...")
		for {
			decoder := json.NewDecoder(resp.Body)
			var ev Event
			err = decoder.Decode(&ev)
			if err != nil {
				errc <- err
			}
			event := ev.Object
			log.Println("new cluster size:", event.Size, event.Metadata)
			events <- event
		}
	}()

	return events, errc
}

func main() {
	c := &etcdClusterController{
		kclient: mustCreateClient(masterHost),
	}
	log.Println("etcd cluster controller starts running...")
	c.Run()
}

func mustCreateClient(host string) *unversioned.Client {
	cfg := &restclient.Config{
		Host:  host,
		QPS:   100,
		Burst: 100,
	}
	c, err := unversioned.New(cfg)
	if err != nil {
		panic(err)
	}
	return c
}

func makeEtcdService(etcdName, clusterName string) *api.Service {
	labels := map[string]string{
		"etcd_node":    etcdName,
		"etcd_cluster": clusterName,
	}
	svc := &api.Service{
		ObjectMeta: api.ObjectMeta{
			Name:   etcdName,
			Labels: labels,
		},
		Spec: api.ServiceSpec{
			Ports: []api.ServicePort{{
				Name:       "server",
				Port:       2380,
				TargetPort: intstr.FromInt(2380),
				Protocol:   api.ProtocolTCP,
			}},
			Selector: labels,
		},
	}
	return svc
}

func makeEtcdPod(etcdName, clusterName string, initialCluster []string) *api.Pod {
	pod := &api.Pod{
		ObjectMeta: api.ObjectMeta{
			Name: etcdName,
			Labels: map[string]string{
				"app":          "etcd",
				"etcd_node":    etcdName,
				"etcd_cluster": clusterName,
			},
		},
		Spec: api.PodSpec{
			Containers: []api.Container{
				{
					Command: []string{
						"/usr/local/bin/etcd",
{'fraction_non_alphanumeric': 0.1543448275862069, 'fraction_numerical': 0.012413793103448275, 'mean_word_length': 3.3032640949554897, 'pattern_counts': {'":': 10, '<': 23, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5019037', 'n_tokens_mistral': 2997, 'n_tokens_neox': 2660, 'n_words': 577}