Extracted from rest-layer

Author: Olivier Poitrey

.travis.yml

@@ -0,0 +1,7 @@
+language: go
+go:
+- 1.6
+- tip
+env:
+  allow_failures:
+    - go: tip

README.md

@@ -0,0 +1,51 @@
+# REST Layer Hystrix storage handler wrapper
+
+[![godoc](http://img.shields.io/badge/godoc-reference-blue.svg?style=flat)](https://godoc.org/github.com/rs/rest-layer-hystrix) [![license](http://img.shields.io/badge/license-MIT-red.svg?style=flat)](https://raw.githubusercontent.com/rs/rest-layer-hystrix/master/LICENSE) [![build](https://img.shields.io/travis/rs/rest-layer-hystrix.svg?style=flat)](https://travis-ci.org/rs/rest-layer-hystrix)
+
+This [REST Layer](https://github.com/rs/rest-layer) resource storage wrapper uses [hystrix-go](github.com/afex/hystrix-go) to add curuit breaker support to any REST Layer resource storage handler.
+
+## Usage
+
+```go
+import "github.com/rs/rest-layer-hystrix"
+```
+
+Wrap existing storage handler with a name that will be used to construct hystrix commands:
+
+```go
+s := restrix.Wrap("myResource", mem.NewHandler())
+```
+
+Use this handler with a resource:
+
+```go
+index.Bind("foo", foo, s, resource.DefaultConf)
+```
+
+Customize the hystrix commands:
+
+```go
+// Configure hystrix commands
+hystrix.Configure(map[string]hystrix.CommandConfig{
+    "posts.Find": {
+        Timeout:               1000,
+        MaxConcurrentRequests: 100,
+        ErrorPercentThreshold: 25,
+    },
+    "posts.Insert": {
+        Timeout:               1000,
+        MaxConcurrentRequests: 50,
+        ErrorPercentThreshold: 25,
+    },
+    ...
+})
+```
+
+Start the metrics stream handler:
+
+```go
+hystrixStreamHandler := hystrix.NewStreamHandler()
+hystrixStreamHandler.Start()
+log.Print("Serving Hystrix metrics on http://localhost:8081")
+go http.ListenAndServe(net.JoinHostPort("", "8081"), hystrixStreamHandler)
+```

examples/hystrix/main.go

@@ -0,0 +1,117 @@
+package main
+
+import (
+	"log"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+
+	"github.com/afex/hystrix-go/hystrix"
+	"github.com/rs/rest-layer-hystrix"
+	"github.com/rs/rest-layer-mem"
+	"github.com/rs/rest-layer/resource"
+	"github.com/rs/rest-layer/rest"
+	"github.com/rs/rest-layer/schema"
+	"github.com/rs/xaccess"
+	"github.com/rs/xhandler"
+	"github.com/rs/xlog"
+)
+
+var (
+	post = schema.Schema{
+		Fields: schema.Fields{
+			"id":      schema.IDField,
+			"created": schema.CreatedField,
+			"updated": schema.UpdatedField,
+			"title":   {},
+			"body":    {},
+		},
+	}
+)
+
+func main() {
+	index := resource.NewIndex()
+
+	index.Bind("posts", post, restrix.Wrap("posts", mem.NewHandler()), resource.Conf{
+		AllowedModes: resource.ReadWrite,
+	})
+
+	// Create API HTTP handler for the resource graph
+	api, err := rest.NewHandler(index)
+	if err != nil {
+		log.Fatalf("Invalid API configuration: %s", err)
+	}
+
+	// Setup logger
+	c := xhandler.Chain{}
+	c.UseC(xlog.NewHandler(xlog.Config{}))
+	c.UseC(xaccess.NewHandler())
+
+	// Bind the API under the root path
+	http.Handle("/", c.Handler(api))
+
+	// Configure hystrix commands
+	hystrix.Configure(map[string]hystrix.CommandConfig{
+		"posts.MultiGet": {
+			Timeout:               500,
+			MaxConcurrentRequests: 200,
+			ErrorPercentThreshold: 25,
+		},
+		"posts.Find": {
+			Timeout:               1000,
+			MaxConcurrentRequests: 100,
+			ErrorPercentThreshold: 25,
+		},
+		"posts.Insert": {
+			Timeout:               1000,
+			MaxConcurrentRequests: 50,
+			ErrorPercentThreshold: 25,
+		},
+		"posts.Update": {
+			Timeout:               1000,
+			MaxConcurrentRequests: 50,
+			ErrorPercentThreshold: 25,
+		},
+		"posts.Delete": {
+			Timeout:               1000,
+			MaxConcurrentRequests: 10,
+			ErrorPercentThreshold: 10,
+		},
+		"posts.Clear": {
+			Timeout:               10000,
+			MaxConcurrentRequests: 5,
+			ErrorPercentThreshold: 10,
+		},
+	})
+
+	// Start the metrics stream handler
+	hystrixStreamHandler := hystrix.NewStreamHandler()
+	hystrixStreamHandler.Start()
+	log.Print("Serving Hystrix metrics on http://localhost:8081")
+	go http.ListenAndServe(net.JoinHostPort("", "8081"), hystrixStreamHandler)
+
+	// Inject some fixtures
+	fixtures := [][]string{
+		[]string{"POST", "/posts", `{"title": "First Post", "body": "This is my first post"}`},
+		[]string{"POST", "/posts", `{"title": "Second Post", "body": "This is my second post"}`},
+		[]string{"POST", "/posts", `{"title": "Third Post", "body": "This is my third post"}`},
+	}
+	for _, fixture := range fixtures {
+		req, err := http.NewRequest(fixture[0], fixture[1], strings.NewReader(fixture[2]))
+		if err != nil {
+			log.Fatal(err)
+		}
+		w := httptest.NewRecorder()
+		api.ServeHTTP(w, req)
+		if w.Code >= 400 {
+			log.Fatalf("Error returned for `%s %s`: %v", fixture[0], fixture[1], w)
+		}
+	}
+
+	// Serve it
+	log.Print("Serving API on http://localhost:8080")
+	if err := http.ListenAndServe(":8080", nil); err != nil {
+		log.Fatal(err)
+	}
+}

hystrix.go

@@ -0,0 +1,125 @@
+// Package restrix is a REST Layer resource storage wrapper to add hystrix support
+// to the underlaying storage handler.
+package restrix
+
+import (
+	"fmt"
+
+	"github.com/afex/hystrix-go/hystrix"
+	"github.com/rs/rest-layer/resource"
+	"golang.org/x/net/context"
+)
+
+type wrapper struct {
+	resource.Storer
+	getCmd    string
+	findCmd   string
+	insertCmd string
+	updateCmd string
+	deleteCmd string
+	clearCmd  string
+}
+
+type mgetWrapper struct {
+	wrapper
+	multiGetCmd string
+}
+
+// Wrap wraps a REST Layer storage handler to add hystrix support to all
+// handler's methods.
+//
+// Hystrix wraps each storage handlers into an hystrix command. One hystrix
+// command is created per backend actions with the format <name>.<Action>.
+//
+// Actions are Find, Insert, Update, Delete, Clear and MultiGet for handlers
+// implementing MultiGetter interface.
+//
+// You must configure hystrix for each command you want to control and start the
+// stream handler.
+// See https://godoc.org/github.com/afex/hystrix-go/hystrix for more info and
+// examples/hystrix/main.go for a usage example.
+func Wrap(name string, h resource.Storer) resource.Storer {
+	w := wrapper{
+		Storer:    h,
+		getCmd:    fmt.Sprintf("%s.Get", name),
+		findCmd:   fmt.Sprintf("%s.Find", name),
+		insertCmd: fmt.Sprintf("%s.Insert", name),
+		updateCmd: fmt.Sprintf("%s.Update", name),
+		deleteCmd: fmt.Sprintf("%s.Delete", name),
+		clearCmd:  fmt.Sprintf("%s.Clear", name),
+	}
+	if _, ok := h.(resource.MultiGetter); ok {
+		return mgetWrapper{
+			wrapper:     w,
+			multiGetCmd: fmt.Sprintf("%s.MultiGet", name),
+		}
+	}
+	return w
+}
+
+func (w wrapper) Insert(ctx context.Context, items []*resource.Item) error {
+	return hystrix.Do(w.insertCmd, func() error {
+		return w.Storer.Insert(ctx, items)
+	}, nil)
+}
+
+func (w wrapper) Update(ctx context.Context, item *resource.Item, original *resource.Item) error {
+	return hystrix.Do(w.updateCmd, func() error {
+		return w.Storer.Update(ctx, item, original)
+	}, nil)
+}
+
+func (w wrapper) Delete(ctx context.Context, item *resource.Item) error {
+	return hystrix.Do(w.deleteCmd, func() error {
+		return w.Storer.Delete(ctx, item)
+	}, nil)
+}
+
+func (w wrapper) Clear(ctx context.Context, lookup *resource.Lookup) (deleted int, err error) {
+	out := make(chan int, 1)
+	errs := hystrix.Go(w.clearCmd, func() error {
+		deleted, err := w.Storer.Clear(ctx, lookup)
+		if err == nil {
+			out <- deleted
+		}
+		return err
+	}, nil)
+	select {
+	case deleted = <-out:
+	case err = <-errs:
+	}
+	return
+}
+
+func (w wrapper) Find(ctx context.Context, lookup *resource.Lookup, page, perPage int) (list *resource.ItemList, err error) {
+	out := make(chan *resource.ItemList, 1)
+	errs := hystrix.Go(w.findCmd, func() error {
+		list, err := w.Storer.Find(ctx, lookup, page, perPage)
+		if err == nil {
+			out <- list
+		}
+		return err
+	}, nil)
+	select {
+	case list = <-out:
+	case err = <-errs:
+	}
+	return
+}
+
+func (w mgetWrapper) MultiGet(ctx context.Context, ids []interface{}) (items []*resource.Item, err error) {
+	out := make(chan []*resource.Item, 1)
+	errs := hystrix.Go(w.multiGetCmd, func() error {
+		mg := w.wrapper.Storer.(resource.MultiGetter)
+		items, err := mg.MultiGet(ctx, ids)
+		if err == nil {
+			out <- items
+		}
+		return err
+	}, nil)
+	select {
+	case items = <-out:
+	case err = <-errs:
+	}
+	return
+}