Dev/rsearch

rsearch/.gitignore

@@ -0,0 +1 @@
+vendor/

rsearch/TODO

@@ -0,0 +1,3 @@
+1. When starting server read through namespace list and fire up ns.Producer per namespace.
+   Read throug resource list per namespace and send them to Processor
+2. Import Kubernetes and use Objects from there

rsearch/bin/.gitignore

@@ -0,0 +1 @@
+bin

rsearch/bin/main.go

@@ -0,0 +1,57 @@
+package main
+
+import (
+	"flag"
+	"fmt"
+	search "github.com/romana/contrib/rsearch"
+	"log"
+	//	"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 host = flag.String("h", "", "Protocol://host for client to connect to")
+	var searchTag = flag.String("r", "", "Search resources by tag")
+	flag.Parse()
+
+	done := make(chan search.Done)
+
+	config, err := search.NewConfig(*cfgFile)
+	if *host != "" {
+		config.Server.Host = *host
+	}
+
+	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, config)
+		if err != nil {
+			log.Fatal("Namespace watcher failed to start", err)
+		}
+
+		events := search.Conductor(nsEvents, done, config)
+		req := search.Process(events, done, config)
+		log.Println("All routines started")
+		search.Serve(config, req)
+	} else if len(*searchTag) > 0 {
+		if config.Server.Debug {
+			fmt.Println("Making request t the server")
+		}
+		r := search.SearchResource(config, search.SearchRequest{Tag: *searchTag})
+		response, _ := json.Marshal(r)
+		if err != nil {
+			panic(err)
+		}
+		fmt.Println(string(response))
+	}
+
+}

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,47 @@
+package rsearch
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"log"
+	"net/http"
+)
+
+// SearchResource connects to instance of a server
+// and resolves SearchRequest
+func SearchResource(config Config, req SearchRequest) SearchResponse {
+	// TODO need to make url configurable
+	url := config.Server.Host + ":" + config.Server.Port
+	data := []byte(`{ "tag" : "` + req.Tag + `"}`)
+	if config.Server.Debug {
+		log.Println("Making request with", string(data))
+	}
+
+	// Make request
+	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 {
+		log.Println("HTTP request failed", url, err)
+		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 {
+		log.Println("Failed to decode", response.Body)
+		panic(err)
+	}
+
+	if config.Server.Debug {
+		fmt.Println("Decoded response form a server", sr)
+	}
+	return sr
+}

rsearch/conductor.go

@@ -0,0 +1,43 @@
+package rsearch
+
+func manageResources(ns Event, terminators map[string]chan Done, config Config, out chan Event) {
+	uid := ns.Object.Metadata.Uid
+	if ns.Type == "ADDED" {
+		done := make(chan Done)
+		terminators[uid] = done
+		ns.Object.Produce(out, terminators[uid], config)
+	} else if ns.Type == "DELETED" {
+		close(terminators[uid])
+		delete(terminators, uid)
+	} else if ns.Type == "_CRASH" {
+		for uid, c := range terminators {
+			close(c)
+			delete(terminators, uid)
+		}
+	}
+}
+
+// Conductor manages a set of goroutines one per namespace.
+func Conductor(in <-chan Event, done <-chan Done, config Config) <-chan Event {
+	// Idea of this map is to keep termination channels organized
+	// so when DELETED event occurs on a namespace it would be possible
+	// to terminater related goroutine
+	var terminators map[string]chan Done
+	terminators = make(map[string]chan Done)
+
+	ns := Event{}
+	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,48 @@
+package rsearch
+
+import (
+	"gopkg.in/gcfg.v1"
+)
+
+// Done is an alias for empty struct, used for termination channels
+type Done struct{}
+
+// Config is a top level struct describing expected structure of config file.
+type Config struct {
+	Resource Resource
+	Server   Server
+	Api      Api
+}
+
+// Server is a config section describing server instance of this package.
+type Server struct {
+	Port  string
+	Host  string
+	Debug bool
+}
+
+// API is a config section describing kubernetes API parameters.
+type Api struct {
+	Url          string
+	NamespaceUrl string
+}
+
+// Resource is a config section describing kubernetes resource to be cached and searched for.
+type Resource struct {
+	Name       string
+	Type       string
+	Selector   string
+	Namespaced string
+	UrlPrefix  string
+	UrlPostfix string
+}
+
+// NewConfig parsing config file and returning initialized instance of Config structure
+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/doc.go

@@ -0,0 +1,3 @@
+// This package prvides means of searching kubernets objects by their selectors
+// the goal is to give the answer to a question - which resources are selecting pods with given label
+package rsearch

rsearch/processor.go

@@ -0,0 +1,97 @@
+package rsearch
+
+import (
+	"log"
+)
+
+/*
+{"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"}}}}
+*/
+
+// Process is a goroutine that consumes resource update events and maintains a searchable
+// cache of all known resources. It also accepts search requests and perform searches.
+func Process(in <-chan Event, done chan Done, config Config) chan<- SearchRequest {
+	// Channel to submit SearchRequest's into
+	req := make(chan SearchRequest)
+
+	// storage map is a cache of known KubeObjects
+	// arranged by NPid
+	storage := make(map[string]KubeObject)
+
+	// search map is a cache of known NPid's
+	// arranged by Selectors, where selector being
+	// a field by which we search
+	search := make(map[string]map[string]bool)
+
+	go func() {
+		for {
+			select {
+			case e := <-in:
+				// On incoming event update caches
+				updateStorage(e, storage, search, config)
+			case request := <-req:
+				// On incoming search request return a list
+				// of resources with matching Selectors
+				processSearchRequest(storage, search, request, config)
+			case <-done:
+				return
+			}
+		}
+	}()
+
+	return req
+}
+
+func processSearchRequest(storage map[string]KubeObject, search map[string]map[string]bool, req SearchRequest, config Config) SearchResponse {
+	if config.Server.Debug {
+		log.Println("Received request", req)
+	}
+
+	var resp []KubeObject
+
+	if config.Server.Debug {
+		log.Printf("Index map has following %s, request tag is %s ", search, req.Tag)
+	}
+
+	// Assembling response.
+	for NPid, _ := range search[string(req.Tag)] {
+		if config.Server.Debug {
+			log.Printf("Assembling response adding %s to %s", resp, storage[NPid])
+		}
+		resp = append(resp, storage[NPid])
+	}
+
+	if config.Server.Debug {
+		log.Printf("Dispatching final response %s", resp)
+	}
+
+	req.Resp <- resp // TODO see if it may hang up here
+	return resp
+}
+
+func updateStorage(e Event, storage map[string]KubeObject, search map[string]map[string]bool, config Config) {
+	NPid := e.Object.makeId()
+	Selector := e.Object.getSelector(config)
+
+	if e.Type == "ADDED" {
+		if config.Server.Debug {
+			log.Printf("Processing ADD request for %s", e.Object.Metadata.Name)
+		}
+		storage[NPid] = e.Object
+		if _, ok := search[Selector]; !ok {
+			m := make(map[string]bool)
+			search[Selector] = m
+		}
+		search[Selector][NPid] = true
+	} else if e.Type == "DELETED" {
+		if config.Server.Debug {
+			log.Printf("Processing DELETE request for %s", e.Object.Metadata.Name)
+		}
+		delete(storage, NPid)
+		delete(search[Selector], NPid)
+	} else {
+		if config.Server.Debug {
+			log.Printf("Received unindentified request %s for %s", e.Type, e.Object.Metadata.Name)
+		}
+	}
+}