[WIP] Kubernetes Service

main.go

@@ -1,11 +1,15 @@
 package main
+//import _ "net/http/pprof"
 
 import (
 	"flag"
 	"log"
 	"net/url"
 	"strings"
 	"time"
+        "runtime/debug"
+        //"net/http"
+        "runtime"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/session"
@@ -21,9 +25,10 @@ var (
 
 	manualPeers = flag.String("peers", "", "CSV separated list of peers' URLs")
 	srvDNSName  = flag.String("peer-srv-endpoint", "", "SRV record prefix for peer discovery (intended for use with kubernetes headless services)")
+	srvPortName  = flag.String("peer-srv-port-name", "", "SRV record prefix for peer discovery (intended for use with kubernetes headless services)")
 	u           Updater
 
-	bucket = flag.String("bucket", "", "Bucket ot use for S3 client")
+	bucket = flag.String("bucket", "", "Bucket to use for S3 client")
 )
 
 func parseArgs() {
@@ -41,6 +46,16 @@ func parseArgs() {
 		log.Fatal("-peers & -peer-srv-endpoint are mututally exclusive options")
 	}
 
+	if *srvDNSName == "" {
+		if *srvPortName != "" {
+			log.Fatal("-peer-srv-endpoint & -peer-srv-port-name must be set together")
+		}
+	} else {
+		if *srvPortName == "" {
+			log.Fatal("-peer-srv-endpoint & -peer-srv-port-name must be set together")
+		}
+	}
+
 	if peers := strings.Split(*manualPeers, ","); len(peers) > 0 {
 		for _, p := range peers {
 			_, err := url.Parse(p)
@@ -56,15 +71,27 @@ func parseArgs() {
 
 		u = StaticPeers(*self, append(peers, *self))
 	case *srvDNSName != "":
-		u = SRVDiscoveredPeers(*self, *srvDNSName, time.Second*15)
+		u = SRVDiscoveredPeers(*self, *srvDNSName, *srvPortName, time.Second*15)
 	default:
 		log.Fatal("must set a peer option: -peers || -peer-srv-endpoint")
 	}
 }
 
 func main() {
+        debug.SetGCPercent(5)
+        ticker := time.NewTicker(time.Millisecond * 300000)
+        go func() {
+           for t := range ticker.C {
+               log.Println("Forcing GC", t)
+               runtime.GC()
+           }
+        }()
 	parseArgs()
-
+        /*
+        go func() {
+	   log.Println(http.ListenAndServe("localhost:6060", nil))
+        }()
+        */
 	s3m := NewS3Manager(
 		s3.New(session.Must(session.NewSession(&aws.Config{
 			Region: aws.String("us-west-2"),
@@ -75,6 +102,6 @@ func main() {
 	)
 
 	cs := newCacheServer(s3m, *self, groupcache.GetterFunc(s3m.Getter), u)
-
+       
 	graceful.Run(*bind, time.Second*15, cs)
 }

peers.go

@@ -43,8 +43,8 @@ func StaticPeers(self string, peers []string) Updater {
 	}
 }
 
-func srvLookup(srvName string) ([]string, error) {
-	cname, targets, err := net.LookupSRV("bazelcache", "tcp", srvName)
+func srvLookup(srvName string, srvPortName string) ([]string, error) {
+	cname, targets, err := net.LookupSRV(srvPortName, "tcp", srvName)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to resolve SRV record")
 	}
@@ -53,9 +53,18 @@ func srvLookup(srvName string) ([]string, error) {
 	// Build peer list from SRV targets
 	peers := make([]string, len(targets))
 	for i, addr := range targets {
+		ip, err := net.LookupHost(addr.Target)
+		if err != nil {
+			return nil, errors.Wrap(err, "failed to resolve SRV record")
+		}
+
+		if len(ip) != 1 {
+			return nil, errors.Errorf("Got multiple hosts for ip: %q in %q", addr.Target, ip)
+		}
+
 		peers[i] = (&url.URL{
 			Scheme: "http",
-			Host:   net.JoinHostPort(addr.Target, strconv.Itoa(int(addr.Port))),
+			Host:   net.JoinHostPort(ip[0], strconv.Itoa(int(addr.Port))),
 		}).String()
 		log.Printf("SRV peer: %q", peers[i])
 	}
@@ -64,15 +73,17 @@ func srvLookup(srvName string) ([]string, error) {
 }
 
 // SRVDiscoveredPeers periodically sends SRV requests to the provided DNS name to discover (& set) the pool's peers
-func SRVDiscoveredPeers(self string, srvPeerDNSName string, updateInterval time.Duration) Updater {
+func SRVDiscoveredPeers(self string, srvPeerDNSName string, srvPortName string, updateInterval time.Duration) Updater {
 	update := func(pool *groupcache.HTTPPool) error {
-		peers, err := srvLookup(srvPeerDNSName)
+		peers, err := srvLookup(srvPeerDNSName, srvPortName)
 		if err != nil {
-			return errors.Wrap(err, "srv lookup failed")
+			pool.Set(self)
+			return nil
+			// return errors.Wrap(err, "srv lookup failed")
 		}
 
 		if !selfInPeers(self, peers) {
-			return errors.Errorf("self not in peers: %q not in %q", self, peers)
+			// return errors.Errorf("self not in peers: %q not in %q", self, peers)
 		}
 
 		pool.Set(peers...)

server.go

@@ -36,7 +36,7 @@ func newCacheServer(s3m *S3Manager, self string, getter groupcache.Getter, updat
 	// Create group of cached objects
 	group := groupcache.NewGroup(
 		"bazelcache",
-		2<<32,
+		2<<34,
 		getter,
 	)
 	go logCacheStats(group, time.Second*15)
@@ -68,11 +68,11 @@ func (c *cacheServer) ServeHTTP(rw http.ResponseWriter, r *http.Request) {
 
 	m.Handle("/_groupcache/*", c.gpool)
 
-	m.Get("/ac/*", c.handleGET())
-	m.Get("/cas/*", c.handleGET())
+	m.Get("/cache/ac/*", c.handleGET())
+	m.Get("/cache/cas/*", c.handleGET())
 
-	m.Put("/ac/*", c.handlePUT())
-	m.Put("/cas/*", c.handlePUT())
+	m.Put("/cache/ac/*", c.handlePUT())
+	m.Put("/cache/cas/*", c.handlePUT())
 
 	m.ServeHTTP(rw, r)
 }