Dev/rsearch

rsearch/bin/main.go

@@ -0,0 +1,49 @@
+package main
+
+import (
+	"github.com/romana/search"	
+	"flag"
+	"log"
+	"fmt"
+//	"net/http"
+//    "encoding/json"
+
+//	"io"
+)
+
+
+func main() {
+	var cfgFile = flag.String("c", "", "Kubernetes reverse search config file")
+	var server = flag.Bool("s", false, "Start a server")
+	var searchTag = flag.String("r", "", "Search resources by tag")
+	flag.Parse()
+
+	done := make(chan search.Done)
+
+	config, err := search.NewConfig(*cfgFile); 
+	if err != nil {
+		fmt.Printf("Can not read config file %s, %s\n", *cfgFile, err)
+		return
+	}
+
+	if *server {
+		fmt.Println("Starting server")
+		nsUrl := fmt.Sprintf("%s/%s", config.Api.Url, config.Api.NamespaceUrl)
+		nsEvents, err :=  search.NsWatch(done, nsUrl)
+		if err != nil {
+			log.Fatal("Namespace watcher failed to start", err)
+		}
+
+		events := search.Conductor(nsEvents, done, config)
+		req, resp := search.Process(events, done, config)
+		log.Println("All routines started")
+		search.Serve(config, req, resp)
+	} else if len(*searchTag) > 0 {
+		if config.Server.Debug {
+			fmt.Println("Making request t the server")
+		}
+		r := search.SearchResource(config, search.SearchRequest{*searchTag})
+		fmt.Println(r)
+	}
+
+}

rsearch/build.sh

@@ -0,0 +1,5 @@
+#!/bin/bash
+
+find . -type d | while read line; do
+	( echo Building in the $line ....; cd $line && go build .)
+done

rsearch/client.go

@@ -0,0 +1,42 @@
+package search
+
+import (
+	"net/http"
+	"encoding/json"
+	"fmt"
+	"log"
+	"bytes"
+)
+
+func SearchResource(config Config, req SearchRequest) SearchResponse {
+	url := "http://localhost:" + config.Server.Port
+	data := []byte(`{ "tag" : "` + req.Tag + `"}`)
+	if config.Server.Debug {
+		log.Println("Making request with", string(data))
+	}
+
+	request, err := http.NewRequest("POST", url, bytes.NewBuffer(data))
+	request.Header.Set("Content-Type", "application/json")
+	client := &http.Client{}
+	response, err := client.Do(request)
+	if err != nil {
+		panic(err)
+	}
+	defer response.Body.Close()
+	decoder := json.NewDecoder(response.Body)
+
+	sr := SearchResponse{}
+
+	if config.Server.Debug {
+		log.Println("Trying to decode", response.Body)
+	}
+	err = decoder.Decode(&sr)
+	if err != nil {
+		panic(err)
+	}
+
+	if config.Server.Debug {
+		fmt.Println("Got response form a server", sr)
+	}
+	return sr
+}

rsearch/conductor.go

@@ -0,0 +1,35 @@
+// Conductor is a goroutine that consumes NsEvents
+// and maintains a proper number of Resource goroutines
+package search
+
+func manageResources(ns NsEvent, terminators map[string]chan Done, config Config, out chan Event) {
+	if ns.Type == "ADDED" {
+		done := make(chan Done)
+		terminators[ns.Object.Metadata.Uid] = done
+		ns.Produce(out, terminators[ns.Object.Metadata.Uid], config)
+	} else if ns.Type == "DELETED" {
+		close(terminators[ns.Object.Metadata.Uid])
+		delete(terminators, ns.Object.Metadata.Uid)
+	}
+}
+
+func Conductor(in <-chan NsEvent, done <-chan Done, config Config) <-chan Event {
+	var terminators map[string]chan Done
+	terminators = make(map[string]chan Done)
+
+	ns := NsEvent{}
+	out := make(chan Event)
+
+	go func () {
+		for {
+			select {
+				case ns = <- in:
+					manageResources(ns, terminators, config, out)
+				case <-done:
+					return
+			}
+		}
+	}()
+
+	return out
+}

rsearch/config.go

@@ -0,0 +1,41 @@
+package search
+
+import (
+	"gopkg.in/gcfg.v1"
+)
+
+type Done struct {}
+
+type Config struct {
+	Resource Resource
+	Server Server
+	Api Api
+}
+
+type Server struct {
+	Port string
+	Debug bool
+}
+
+type Api struct {
+	Url string
+	NamespaceUrl string
+}
+
+type Resource struct {
+	Name string
+	Type string
+	Selector string
+	Namespaced string
+	UrlPrefix string
+	UrlPostfix string
+}
+
+func NewConfig(configFile string) (Config, error) {
+	cfg := Config{}
+	if err := gcfg.ReadFileInto(&cfg, configFile); err != nil {
+		return cfg, err
+	}
+
+	return cfg, nil
+}

rsearch/config.ini

@@ -0,0 +1,18 @@
+; comment
+[api]
+url=http://192.168.0.10:8080
+namespaceUrl=api/v1/namespaces/?watch=true
+
+[resource]
+;type=builtin # TBD
+type=3rdParty
+;namespaced=false # TBD
+namespaced=true
+urlPrefix=apis/romana.io/demo/v1/namespaces
+urlPostfix=networkpolicys/?watch=true
+name=NetworkPolicy
+selector=podSelector
+
+[server]
+port=9700
+debug=false

rsearch/namespace.go

@@ -0,0 +1,79 @@
+package search
+
+import (
+	"fmt"
+	"log"
+	"time"
+	"net/http"
+	"encoding/json"
+)
+
+/*
+{"type":"ADDED","object":{"kind":"Namespace","apiVersion":"v1","metadata":{"name":"default","selfLink":"/api/v1/namespaces/default","uid":"d10db271-dc03-11e5-9c86-0213e1312dc5","resourceVersion":"6","creationTimestamp":"2016-02-25T21:07:45Z"},"spec":{"finalizers":["kubernetes"]},"status":{"phase":"Active"}}}
+*/
+
+type NsEvent struct {
+	Type string	`json:"Type"`
+	Object NsObject	`json:"object"`
+}
+
+type NsObject struct {
+	Kind	string	`json:"kind"`
+	Spec	Spec	`json:"spec"`
+	ApiVersion	string	`json:"apiVersion"`
+	Metadata	Metadata	`json:"metadata"`
+	Status	map[string]string	`json:"status"`
+}
+
+func NsWatch(done <-chan Done,  url string) (<-chan NsEvent, error) {
+	out := make(chan NsEvent)
+	resp, err := http.Get(url)
+	if err != nil {
+		return out, err
+	}
+	tick := time.Tick(1 * time.Second)
+	fmt.Println(resp.Body)
+	dec := json.NewDecoder(resp.Body)
+	var e NsEvent
+
+	go func() {
+		for {
+			select {
+			case <- tick:
+				dec.Decode(&e)
+				out <- e
+			case <- done:
+				return
+			}
+		}
+	}()
+
+	return out, nil
+}
+
+func (ns NsEvent) Produce(out chan Event, done <-chan Done, config  Config) error {
+	url := fmt.Sprintf("%s/%s/%s/%s", config.Api.Url, config.Resource.UrlPrefix, ns.Object.Metadata.Name, config.Resource.UrlPostfix)
+	log.Println("Launching producer to listen on ", url)
+	tick := time.Tick(1 * time.Second)
+
+	resp, err  := http.Get(url)
+	if err != nil {
+		return  err
+	}
+
+	dec := json.NewDecoder(resp.Body)
+	var e Event
+	go func() {
+		for {
+			select {
+			case <- tick:
+				dec.Decode(&e)
+				out <- e
+			case <- done:
+				return
+			}
+		}
+	}()
+
+	return nil
+}

rsearch/namespace_test.go

@@ -0,0 +1,90 @@
+package search
+
+import (
+	"net/http"
+	"gopkg.in/gcfg.v1"
+	"testing"
+	"math/rand"
+	"time"
+	"log"
+	"fmt"
+)
+
+const cfgStr = `
+; comment
+[api]
+url=http://127.0.0.1
+namespaceUrl=api/v1/namespaces
+
+[resource]
+type=3rdParty
+namespaced=true
+urlPrefix=apis/romana.io/demo/v1/namespaces
+urlPostfix=networkpolicys
+name=NetworkPolicy
+selector=podSelector
+
+[server]
+port=9700
+debug=false
+`
+
+const testNs = `{"type":"ADDED","object":{"kind":"Namespace","apiVersion":"v1","metadata":{"name":"default","selfLink":"/api/v1/namespaces/default","uid":"d10db271-dc03-11e5-9c86-0213e1312dc5","resourceVersion":"6","creationTimestamp":"2016-02-25T21:07:45Z"},"spec":{"finalizers":["kubernetes"]},"status":{"phase":"Active"}}}
+`
+
+const testPolicy = `{"type":"ADDED","object":{"apiVersion":"romana.io/demo/v1","kind":"NetworkPolicy","metadata":{"name":"pol1","namespace":"default","selfLink":"/apis/romana.io/demo/v1/namespaces/default/networkpolicys/pol1","uid":"d7036130-e119-11e5-aab8-0213e1312dc5","resourceVersion":"119875","creationTimestamp":"2016-03-03T08:28:00Z","labels":{"owner":"t1"}},"spec":{"allowIncoming":{"from":[{"pods":{"tier":"frontend"}}],"toPorts":[{"port":80,"protocol":"TCP"}]},"podSelector":{"tier":"backend"}}}}
+`
+
+func fakeNsHandler(w http.ResponseWriter, req *http.Request) {
+	fmt.Fprint(w, testNs)
+}
+
+func fakePolicyHandler(w http.ResponseWriter, req *http.Request) {
+	fmt.Fprint(w, testPolicy)
+}
+
+func fakeServer(config Config) {
+	http.HandleFunc("/" + config.Api.NamespaceUrl, fakeNsHandler)
+	policyUrl := "/" + config.Resource.UrlPrefix + "/default/" + config.Resource.UrlPostfix
+	http.HandleFunc(policyUrl, fakePolicyHandler)
+	log.Fatal(http.ListenAndServe(config.Api.Url[7:], nil))
+}
+
+// Testing ability of nswatch to watch ns events
+func TestNsWatch(t *testing.T) {
+	config := Config{}
+	err := gcfg.ReadStringInto(&config, cfgStr)
+	if err != nil {
+		t.Errorf("Failed to parse gcfg data: %s", err)
+	}
+
+	r := rand.New(rand.NewSource(time.Now().UnixNano()))
+	port := 50000 + r.Intn(1000)
+	config.Api.Url = fmt.Sprintf("%s:%d", config.Api.Url, port)
+
+	go fakeServer(config)
+	time.Sleep(time.Duration(5 * time.Second))
+
+	done := make(chan Done)
+	url := config.Api.Url + "/" + config.Api.NamespaceUrl
+	nsEvents, err :=  NsWatch(done, url)
+	if err != nil {
+		t.Error("Namespace watcher failed to start", err)
+	}
+
+	ns := <- nsEvents
+	if ns.Object.Metadata.Name != "default" {
+		t.Error("Expecting namespace name = default, got =", ns.Object.Metadata.Name)
+	}
+
+	events := make(chan Event)
+	err = ns.Produce(events, done, config)
+	if err != nil {
+		t.Error("Namespace producer failed to start", err)
+	}
+
+	e := <- events
+	if e.Object.Metadata.Name != "pol1" {
+		t.Error("Expecting policy name = pol1, got =", e.Object.Metadata.Name)
+	}
+}