Update goruntime to latest, 0.2.5. Add new config for watching change…

…s in runtime config folder directly instead of the runtime root dir. (#151)

Signed-off-by: Yuki Sawa <[email protected]>

README.md

@@ -320,6 +320,12 @@ RUNTIME_IGNOREDOTFILES default:"false"
 
 **Configuration files are loaded from RUNTIME_ROOT/RUNTIME_SUBDIRECTORY/config/\*.yaml**
 
+There are two methods for triggering a configuration reload:
+1. Symlink RUNTIME_ROOT to a different directory.
+2. Update the contents inside `RUNTIME_ROOT/RUNTIME_SUBDIRECTORY/config/` directly.
+
+The former is the default behavior. To use the latter method, set the `RUNTIME_WATCH_ROOT` environment variable to `false`. 
+
 For more information on how runtime works you can read its [README](https://github.com/lyft/goruntime).
 
 # Request Fields

go.mod

@@ -13,7 +13,7 @@ require (
 	github.com/gorilla/mux v1.7.4-0.20191121170500-49c01487a141
 	github.com/kavu/go_reuseport v1.2.0
 	github.com/kelseyhightower/envconfig v1.1.0
-	github.com/lyft/goruntime v0.2.1
+	github.com/lyft/goruntime v0.2.5
 	github.com/lyft/gostats v0.4.0
 	github.com/lyft/protoc-gen-validate v0.0.7-0.20180626203901-f9d2b11e4414 // indirect
 	github.com/mediocregopher/radix/v3 v3.5.1

go.sum

@@ -43,8 +43,8 @@ github.com/kelseyhightower/envconfig v1.1.0 h1:4htXR8ameS6KBfrNBoqEgpg0IK2D6rozN
 github.com/kelseyhightower/envconfig v1.1.0/go.mod h1:cccZRl6mQpaq41TPp5QxidR+Sa3axMbJDNb//FQX6Gg=
 github.com/kisielk/errcheck v1.2.0/go.mod h1:/BMXB+zMLi60iA8Vv6Ksmxu/1UDYcXs4uQLJ+jE2L00=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
-github.com/lyft/goruntime v0.2.1 h1:7DebA8oMVuoQ5TQ0j1xR/X2xRagbGrm0e2SoMdt5tRs=
-github.com/lyft/goruntime v0.2.1/go.mod h1:8rUh5gwIPQtyIkIXHbLN1j45HOb8cMgDhrw5GA7DF4g=
+github.com/lyft/goruntime v0.2.5 h1:yRmwOXl3Zns3+Z03fDMWt5+p609rfhIErh7HYCayODg=
+github.com/lyft/goruntime v0.2.5/go.mod h1:8rUh5gwIPQtyIkIXHbLN1j45HOb8cMgDhrw5GA7DF4g=
 github.com/lyft/gostats v0.4.0 h1:PbRWmwidTPk6Y80S6itBWDa+XVt1hGvqFM88TBJYdOo=
 github.com/lyft/gostats v0.4.0/go.mod h1:Tpx2xRzz4t+T2Tx0xdVgIoBdR2UMVz+dKnE3X01XSd8=
 github.com/lyft/protoc-gen-validate v0.0.7-0.20180626203901-f9d2b11e4414 h1:kLCSHuk3X+SI8Up26wM71id7jz77B3zCZDp01UWMVbM=

src/server/server_impl.go

@@ -7,6 +7,7 @@ import (
 	"io"
 	"net/http"
 	"net/http/pprof"
+	"path/filepath"
 	"sort"
 
 	"os"
@@ -187,12 +188,22 @@ func newServer(name string, store stats.Store, localCache *freecache.Cache, opts
 		loaderOpts = append(loaderOpts, loader.AllowDotFiles)
 	}
 
-	ret.runtime = loader.New(
-		s.RuntimePath,
-		s.RuntimeSubdirectory,
-		ret.store.Scope("runtime"),
-		&loader.SymlinkRefresher{RuntimePath: s.RuntimePath},
-		loaderOpts...)
+	if s.RuntimeWatchRoot {
+		ret.runtime = loader.New(
+			s.RuntimePath,
+			s.RuntimeSubdirectory,
+			ret.store.Scope("runtime"),
+			&loader.SymlinkRefresher{RuntimePath: s.RuntimePath},
+			loaderOpts...)
+
+	} else {
+		ret.runtime = loader.New(
+			filepath.Join(s.RuntimePath, s.RuntimeSubdirectory),
+			"config",
+			ret.store.Scope("runtime"),
+			&loader.DirectoryRefresher{},
+			loaderOpts...)
+	}
 
 	// setup http router
 	ret.router = mux.NewRouter()

src/service/ratelimit.go

@@ -57,6 +57,7 @@ type service struct {
 	stats              serviceStats
 	rlStatsScope       stats.Scope
 	legacy             *legacyService
+	runtimeWatchRoot   bool
 }
 
 func (this *service) reloadConfig() {
@@ -75,7 +76,7 @@ func (this *service) reloadConfig() {
 	files := []config.RateLimitConfigToLoad{}
 	snapshot := this.runtime.Snapshot()
 	for _, key := range snapshot.Keys() {
-		if !strings.HasPrefix(key, "config.") {
+		if this.runtimeWatchRoot && !strings.HasPrefix(key, "config.") {
 			continue
 		}
 
@@ -176,7 +177,7 @@ func (this *service) GetCurrentConfig() config.RateLimitConfig {
 }
 
 func NewService(runtime loader.IFace, cache limiter.RateLimitCache,
-	configLoader config.RateLimitConfigLoader, stats stats.Scope) RateLimitServiceServer {
+	configLoader config.RateLimitConfigLoader, stats stats.Scope, runtimeWatchRoot bool) RateLimitServiceServer {
 
 	newService := &service{
 		runtime:            runtime,
@@ -187,6 +188,7 @@ func NewService(runtime loader.IFace, cache limiter.RateLimitCache,
 		cache:              cache,
 		stats:              newServiceStats(stats),
 		rlStatsScope:       stats.Scope("rate_limit"),
+		runtimeWatchRoot:   runtimeWatchRoot,
 	}
 	newService.legacy = &legacyService{
 		s:                          newService,

src/service_cmd/runner/runner.go

@@ -60,7 +60,9 @@ func (runner *Runner) Run() {
 			rand.New(limiter.NewLockedSource(time.Now().Unix())),
 			s.ExpirationJitterMaxSeconds),
 		config.NewRateLimitConfigLoaderImpl(),
-		srv.Scope().Scope("service"))
+		srv.Scope().Scope("service"),
+		s.RuntimeWatchRoot,
+	)
 
 	srv.AddDebugHttpEndpoint(
 		"/rlconfig",

src/settings/settings.go

@@ -20,6 +20,7 @@ type Settings struct {
 	RuntimePath                  string        `envconfig:"RUNTIME_ROOT" default:"/srv/runtime_data/current"`
 	RuntimeSubdirectory          string        `envconfig:"RUNTIME_SUBDIRECTORY"`
 	RuntimeIgnoreDotFiles        bool          `envconfig:"RUNTIME_IGNOREDOTFILES" default:"false"`
+	RuntimeWatchRoot             bool          `envconfig:"RUNTIME_WATCH_ROOT" default:"true"`
 	LogLevel                     string        `envconfig:"LOG_LEVEL" default:"WARN"`
 	RedisSocketType              string        `envconfig:"REDIS_SOCKET_TYPE" default:"unix"`
 	RedisUrl                     string        `envconfig:"REDIS_URL" default:"/var/run/nutcracker/ratelimit.sock"`

test/integration/integration_test.go

@@ -6,6 +6,7 @@ import (
 	"bytes"
 	"fmt"
 	"io/ioutil"
+	"io"
 	"math/rand"
 	"net/http"
 	"os"
@@ -67,6 +68,11 @@ func TestBasicAuthConfig(t *testing.T) {
 	t.Run("WithPerSecondRedisAuthWithLocalCache", testBasicConfigAuth("18093", "true", "1000"))
 }
 
+func TestBasicReloadConfig(t *testing.T) {
+	t.Run("BasicWithoutWatchRoot", testBasicConfigWithoutWatchRoot("8095", "false", "0"))
+	t.Run("ReloadWithoutWatchRoot", testBasicConfigReload("8097", "false", "0", "false"))
+}
+
 func testBasicConfigAuthTLS(grpcPort, perSecond string, local_cache_size string) func(*testing.T) {
 	os.Setenv("REDIS_PERSECOND_URL", "localhost:16382")
 	os.Setenv("REDIS_URL", "localhost:16381")
@@ -97,6 +103,28 @@ func testBasicConfigAuth(grpcPort, perSecond string, local_cache_size string) fu
 	return testBasicBaseConfig(grpcPort, perSecond, local_cache_size)
 }
 
+func testBasicConfigWithoutWatchRoot(grpcPort, perSecond string, local_cache_size string) func(*testing.T) {
+	os.Setenv("REDIS_PERSECOND_URL", "localhost:6380")
+	os.Setenv("REDIS_URL", "localhost:6379")
+	os.Setenv("REDIS_AUTH", "")
+	os.Setenv("REDIS_TLS", "false")
+	os.Setenv("REDIS_PERSECOND_AUTH", "")
+	os.Setenv("REDIS_PERSECOND_TLS", "false")
+	os.Setenv("RUNTIME_WATCH_ROOT", "false")
+	return testBasicBaseConfig(grpcPort, perSecond, local_cache_size)
+}
+
+func testBasicConfigReload(grpcPort, perSecond string, local_cache_size, runtimeWatchRoot string) func(*testing.T) {
+	os.Setenv("REDIS_PERSECOND_URL", "localhost:6380")
+	os.Setenv("REDIS_URL", "localhost:6379")
+	os.Setenv("REDIS_AUTH", "")
+	os.Setenv("REDIS_TLS", "false")
+	os.Setenv("REDIS_PERSECOND_AUTH", "")
+	os.Setenv("REDIS_PERSECOND_TLS", "false")
+	os.Setenv("RUNTIME_WATCH_ROOT", runtimeWatchRoot)
+	return testConfigReload(grpcPort, perSecond, local_cache_size)
+}
+
 func getCacheKey(cacheKey string, enableLocalCache bool) string {
 	if enableLocalCache {
 		return cacheKey + "_local"
@@ -456,3 +484,105 @@ func TestBasicConfigLegacy(t *testing.T) {
 		assert.NoError(err)
 	}
 }
+
+func testConfigReload(grpcPort, perSecond string, local_cache_size string) func(*testing.T) {
+	return func(t *testing.T) {
+		os.Setenv("REDIS_PERSECOND", perSecond)
+		os.Setenv("PORT", "8082")
+		os.Setenv("GRPC_PORT", grpcPort)
+		os.Setenv("DEBUG_PORT", "8084")
+		os.Setenv("RUNTIME_ROOT", "runtime/current")
+		os.Setenv("RUNTIME_SUBDIRECTORY", "ratelimit")
+		os.Setenv("REDIS_PERSECOND_SOCKET_TYPE", "tcp")
+		os.Setenv("REDIS_SOCKET_TYPE", "tcp")
+		os.Setenv("LOCAL_CACHE_SIZE_IN_BYTES", local_cache_size)
+		os.Setenv("USE_STATSD", "false")
+
+		local_cache_size_val, _ := strconv.Atoi(local_cache_size)
+		enable_local_cache := local_cache_size_val > 0
+		runner := runner.NewRunner()
+
+		go func() {
+			runner.Run()
+		}()
+
+		// HACK: Wait for the server to come up. Make a hook that we can wait on.
+		time.Sleep(1 * time.Second)
+
+		assert := assert.New(t)
+		conn, err := grpc.Dial(fmt.Sprintf("localhost:%s", grpcPort), grpc.WithInsecure())
+		assert.NoError(err)
+		defer conn.Close()
+		c := pb.NewRateLimitServiceClient(conn)
+
+		response, err := c.ShouldRateLimit(
+			context.Background(),
+			common.NewRateLimitRequest("reload", [][][2]string{{{getCacheKey("block", enable_local_cache), "foo"}}}, 1))
+		assert.Equal(
+			&pb.RateLimitResponse{
+				OverallCode: pb.RateLimitResponse_OK,
+				Statuses:    []*pb.RateLimitResponse_DescriptorStatus{{Code: pb.RateLimitResponse_OK}}},
+			response)
+		assert.NoError(err)
+
+		runner.GetStatsStore().Flush()
+		loadCount1 := runner.GetStatsStore().NewCounter("ratelimit.service.config_load_success").Value()
+
+		// Copy a new file to config folder to test config reload functionality
+		in, err := os.Open("runtime/current/ratelimit/reload.yaml")
+		if err != nil {
+			panic(err)
+		}
+		defer in.Close()
+		out, err := os.Create("runtime/current/ratelimit/config/reload.yaml")
+		if err != nil {
+			panic(err)
+		}
+		defer out.Close()
+		_, err = io.Copy(out, in)
+		if err != nil {
+			panic(err)
+		}
+		err = out.Close()
+		if err != nil {
+			panic(err)
+		}
+
+		// Need to wait for config reload to take place and new descriptors to be loaded.
+		// Shouldn't take more than 5 seconds but wait 120 at most just to be safe.
+		wait := 120
+		reloaded := false
+		loadCount2 := uint64(0)
+
+		for i := 0; i < wait; i++ {
+			time.Sleep(1 * time.Second)
+			runner.GetStatsStore().Flush()
+			loadCount2 = runner.GetStatsStore().NewCounter("ratelimit.service.config_load_success").Value()
+
+			// Check that successful loads count has increased before continuing.
+			if loadCount2 > loadCount1 {
+				reloaded = true
+				break
+			}
+		}
+
+		assert.True(reloaded)
+		assert.Greater(loadCount2, loadCount1)
+
+		response, err = c.ShouldRateLimit(
+			context.Background(),
+			common.NewRateLimitRequest("reload", [][][2]string{{{getCacheKey("key1", enable_local_cache), "foo"}}}, 1))
+		assert.Equal(
+			&pb.RateLimitResponse{
+				OverallCode: pb.RateLimitResponse_OK,
+				Statuses: []*pb.RateLimitResponse_DescriptorStatus{
+					newDescriptorStatus(pb.RateLimitResponse_OK, 50, pb.RateLimitResponse_RateLimit_SECOND, 49)}},
+			response)
+		assert.NoError(err)
+
+		err = os.Remove("runtime/current/ratelimit/config/reload.yaml")
+		if err != nil {
+			panic(err)
+		}
+	}
+}

test/integration/runtime/current/ratelimit/reload.yaml

@@ -0,0 +1,16 @@
+domain: reload
+descriptors:
+  - key: key1
+    rate_limit:
+      unit: second
+      requests_per_unit: 50
+
+  - key: block
+    rate_limit:
+      unit: second
+      requests_per_unit: 0
+
+  - key: one_per_minute
+    rate_limit:
+      unit: minute
+      requests_per_unit: 1

test/service/ratelimit_legacy_test.go

@@ -224,7 +224,7 @@ func TestInitialLoadErrorLegacy(test *testing.T) {
 		func([]config.RateLimitConfigToLoad, stats.Scope) {
 			panic(config.RateLimitConfigError("load error"))
 		})
-	service := ratelimit.NewService(t.runtime, t.cache, t.configLoader, t.statStore)
+	service := ratelimit.NewService(t.runtime, t.cache, t.configLoader, t.statStore, true)
 
 	request := common.NewRateLimitRequestLegacy("test-domain", [][][2]string{{{"hello", "world"}}}, 1)
 	response, err := service.GetLegacyService().ShouldRateLimit(nil, request)

test/service/ratelimit_test.go

@@ -82,7 +82,7 @@ func (this *rateLimitServiceTestSuite) setupBasicService() ratelimit.RateLimitSe
 	this.configLoader.EXPECT().Load(
 		[]config.RateLimitConfigToLoad{{"config.basic_config", "fake_yaml"}},
 		gomock.Any()).Return(this.config)
-	return ratelimit.NewService(this.runtime, this.cache, this.configLoader, this.statStore)
+	return ratelimit.NewService(this.runtime, this.cache, this.configLoader, this.statStore, true)
 }
 
 func TestService(test *testing.T) {
@@ -225,7 +225,7 @@ func TestInitialLoadError(test *testing.T) {
 		func([]config.RateLimitConfigToLoad, stats.Scope) {
 			panic(config.RateLimitConfigError("load error"))
 		})
-	service := ratelimit.NewService(t.runtime, t.cache, t.configLoader, t.statStore)
+	service := ratelimit.NewService(t.runtime, t.cache, t.configLoader, t.statStore, true)
 
 	request := common.NewRateLimitRequest("test-domain", [][][2]string{{{"hello", "world"}}}, 1)
 	response, err := service.ShouldRateLimit(nil, request)