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,65 @@
+package main
+
+import (
+	"encoding/json"
+	"flag"
+	"fmt"
+	"log"
+
+	search "github.com/romana/contrib/rsearch"
+)
+
+func main() {
+	var cfgFile = flag.String("c", "", "Kubernetes reverse search config file")
+	var server = flag.Bool("server-mode", false, "Start a server")
+	var host = flag.String("host", "", "Host for client to connect to")
+	var proto = flag.String("protocol", "", "Protocol to use for client connect to")
+	var port = flag.String("port", "", "TCP port 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 err != nil {
+		log.Fatalf("Can not read config file %s, %s\n", *cfgFile, err)
+	}
+
+	if *host != "" {
+		config.Server.Host = *host
+	}
+
+	if *proto != "" {
+		config.Server.Proto = *proto
+	}
+
+	if *port != "" {
+		config.Server.Port = *port
+	}
+
+	if *server {
+		log.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 {
+			log.Println("Making request to the server")
+		}
+		r := search.SearchResource(config, search.SearchRequest{Tag: *searchTag})
+		response, err := json.Marshal(r)
+		if err != nil {
+			log.Fatal("Failed to parse out server response, ", err)
+		}
+		fmt.Println(string(response))
+	} else {
+		log.Fatal("Either -s or -r must be given")
+	}
+}

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,65 @@
+// Copyright (c) 2016 Pani Networks
+// 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 rsearch
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"log"
+	"net/http"
+)
+
+// SearchResource makes HTTP request to the server
+// instance if this library.
+func SearchResource(config Config, req SearchRequest) SearchResponse {
+	// TODO need to make url configurable
+	url := fmt.Sprintf("%s://%s:%s", config.Server.Proto, config.Server.Host, config.Server.Port)
+	data := []byte(`{ "tag" : "` + req.Tag + `"}`)
+	if config.Server.Debug {
+		log.Printf("Making request with %s to the %s\n", string(data), url)
+	}
+
+	// Making request.
+	request, err := http.NewRequest("POST", url, bytes.NewBuffer(data))
+	if err != nil {
+		log.Fatal("Failed to connect to server", url)
+	}
+
+	request.Header.Set("Content-Type", "application/json")
+	client := &http.Client{}
+	response, err := client.Do(request)
+	if err != nil {
+		log.Fatal("HTTP request failed", url, 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,76 @@
+// Copyright (c) 2016 Pani Networks
+// 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 rsearch
+
+import (
+	"log"
+)
+
+// manageResources manages map of termination channels and fires up new
+// per-namespace gorotines when needed.
+func manageResources(ns Event, terminators map[string]chan Done, config Config, out chan Event) {
+	uid := ns.Object.Metadata.Uid
+	if ns.Type == KubeEventAdded {
+		if _, ok := terminators[uid]; ok {
+			log.Println("Received ADDED event for uid that is already known, ignoring ", uid)
+			return
+		}
+
+		done := make(chan Done)
+		terminators[uid] = done
+		ns.Object.Produce(out, terminators[uid], config)
+	} else if ns.Type == KubeEventDeleted {
+		if _, ok := terminators[uid]; !ok {
+			log.Println("Received DELETED event for uid that is not known, ignoring ", uid)
+			return
+		}
+
+		close(terminators[uid])
+		delete(terminators, uid)
+	} else if ns.Type == InternalEventDeleteAll {
+		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 {
+	// done in arguments is a channel that can be used to stop Conductor itsefl
+	// while map of Done's below is for terminating managed gorotines.
+
+	// 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.
+	terminators := 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,65 @@
+// Copyright (c) 2016 Pani Networks
+// 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 rsearch
+
+import (
+	"gopkg.in/gcfg.v1"
+)
+
+// Done is an alias for empty struct, used to make broadcast channels
+// for terminating goroutines.
+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
+	Proto 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 Config{}, err
+	}
+
+	return cfg, nil
+}

rsearch/config.ini

@@ -0,0 +1,20 @@
+; 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
+proto=http
+host=localhost
+debug=false

rsearch/doc.go

@@ -0,0 +1,18 @@
+// Copyright (c) 2016 Pani Networks
+// 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.
+
+// This package provides 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