prepare 6.7.9 release (#194)

* fix product name in docs * fix the broken images in our repo docs (#217) * in auto-config mode, return 503 for client requests till configuration is complete (#219) * always send stream updates to clients regardless of version checking (#224) * Enable hourly Relay integration tests against Production (#223) * Record response body when relay archive download test fails (#225) There is some additional information provided in the response body for certain 504 error codes and this should allow us to see it. * Fix the broken production integration tests (#226) * improve metrics documentation and fix route strings in docs (#227) * improve metrics documentation and fix route strings in docs * clarify mobile + browser terminology * use more efficient jsonstream encoding/decoding for stream data and evaluations (#228) * recognize alias events from non-v3-schema payloads and forward them unchanged * environment should still be usable even if the client timed out * bump dependency versions for SDK fixes * update go-server-sdk-dynamodb to get newer AWS SDK * minor clarification about Relay.Close (#234) * Use the Go releaser template (#233) * [ch102248] big segment sync with redis (#235) * fix makefile so it tries building all the test code first before running any of it (#239) * (big segments #1) add basic big segments configuration for SDK clients (#237) * (big segments #2) add more abstraction around big segments implementation for testability (#238) * [102253] bigsegment status / config (#242) * big segments integration test + misc fixes (#240) * use latest URL paths for big segments endpoints * add SDK DynamoDB integration for big segments (#241) * fix broken link in Markdown docs (#246) * make sure newly added credentials for existing environments are accepted in requests (#244) * don't return 503 if SDK initialization has timed out * add in-repo docs about error/503 behavior (#249) * [ch102255] BigSegments DynamoDB (#245) * add init timeout config option + better test coverage + misc refactoring (#250) * fix example build command * use public prerelease tags instead of private dependencies * fix Go installation in CI * update SDK dependencies for JSON number parsing bugfix * update gorilla/mux to 1.8.0 * update OpenCensus packages * add Go 1.16 CI + "latest Go" CI + use latest 1.15 patch for release * cimg images use "current", not "latest" * seems there isn't any cimg/go "latest" or "current" * add daily package build test in CI * job names * bump SDK version for traffic allocation feature * [ch113491] update alpine base image (#258) * use latest prerelease SDK * fix enabling of test tags in CI * add DynamoDB docker image in CI * set a polling base URI in end-to-end tests since big segments logic will use it * fix initialization logic so SDK client creation errors aren't lost when big segments are enabled * fix use of prefix key in DynamoDB + improve tests (#260) * more debug logging, less info logging for big segments logic * make logging of big segments patch version mismatch clearer and use Warn level * fix log parameter * fix DynamoDB updates for big segments metadata * add test to make sure sync time and cursor can be updated independently * only start big seg synchronizer if necessary * use SDK GA releases * change applyPatch to exit early on version mismatch; go back to restarting stream in this case * add unit tests for version mismatch behavior + DRY tests * add log assertion * fix retry logic on big segments stream failure * add more logging for big segments connection status * fix logging assertion * add more big segments integration tests * fix overly-time-sensitive file data tests * fix more flaky tests * run big segments tests with DynamoDB too * Migrate transitive dep (jwt-go) to use modern version without vulnerability. * Edit doc * move Relay release logic to .ldrelease script * suppress SDK big segments status query if we've never synced big segments * dump Relay logs including debug logs if integration test fails * include environment prefix in BigSegmentSynchronizer logging * increase big segment integration test timeout (#274) * generate client-side stream pings if big segments have changed * clear big segments cache as needed + simplify state management * fix tests and simplify component creation * use GA releases of SDK packages * disable CI package-build-test in Go 1.16+ * Migrate Relay release to Releaser v2 and support dry run (#278) * Adding degraded doc blurb for big segments (#280) * respect Redis password & TLS options for big segments; add Redis password integration tests * redact Redis URL password in logs and status resource * update go-server-sdk-redis-redigo to 1.2.1 for Redis URL logging fix * Part 1, add the config and the documentation for the new config * Part 2, Add the configuration validation and test * Part 3, the actual logic to include the headers in the CORS Access-Control-Allow-Headers * Linter * update Alpine version to 3.14.2 to fix openssl CVEs * Fix the global variable modification * Go format * turn off unnecessary metrics integrations in config for Docker smoke test * rename test.env to smoke-test.env to clarify what it's for * fix setting of custom Access-Control-Allow-Origin and add test (#285) * add more explanatory test output and more verbose debugging for big segments integration tests (#287) * update to Go 1.16.10 + Alpine 3.14.3; add some docs about releases (#288) * update go-server-sdk-consul version for Consul API version update * override x/crypto dependency version for CVE-2020-29652 * bump Prometheus dependency to eliminate jwt-go vulnerability * drop support for Go 1.14 & 1.15 * make sure defaults are always applied for base URL properties * rm unused * rm unnecessary linter directive * add separate configuration for server-side/client-side SDK base URLs & update the defaults * remove Whitesource CI job + remove obsolete dependency issue note * don't include any big segment status info in status resource unless that feature is active (#296) * don't include any big segment status info in status resource unless that feature is active * fix Big Segments staleness logic in status resource * documentation * update x/text package for vulnerability GO-2021-0113 * add Trivy security scan to CI (#297) * add daily re-scan with Trivy * update Go version to 1.17.6 (#301) * always terminate if auto-config stream fails with a fatal error * pass along tags header when proxying events * comments, rm debugging * fix auth header logic * fix auth header logic some more * comments * add tags header to CORS header whitelist (#304) * update to Alpine 3.14.4 for CVE-2022-0778 fix * force upgrade of openssl in Alpine * also upgrade libretls * fix it in both files * update to Alpine 3.14.5 for CVE-2022-0778/CVE-2018-25032 (#308) * update to Alpine 3.14.5 for CVE-2022-0778 * revert patches that are now included in Alpine 3.14.5 * add scripts for checking and updating Go/Alpine versions (#309) * update to Alpine 3.14.5 for CVE-2022-0778 * add scripts for checking and updating Go/Alpine versions * also make sure the Docker images really exist * update CONTRIBUTING.md * fix file rename * revert patches that are now included in Alpine 3.14.5 * update Alpine to 3.14.6 for CVE-2022-28391 * update SDK packages (includes sc-136333 fix) * don't include "v" prefix in Docker image version * update go-server-sdk-dynamodb for data size error fix & add docs (#316) * update builds to use Go 1.17.9 and fix the update script * update go-server-sdk-consul to latest release * update remote Docker version * update golang.org/x/crypto for CVE-2022-27191 (#321) * update golang.org/x/crypto for CVE-2022-27191 * fix go.sum * update eventsource for SSE output efficiency fix (#322) * Cache the replay event in case we get multiple new client connections (#189) * Cache the replay event in case we get multiple new client connections * Use singleflight to ensure only one replay event is generated at a time Co-authored-by: Moshe Good <[email protected]> * don't install curl in Docker images * fix makefile logic for lint step * remove indirect curl-based request logic in integration tests * fix linter installation * update Go to 1.17.11, Alpine to 3.16.0 * improve concurrency test to verify that the data is or isn't from a separate query * fix lint warnings and remove unnecessary error return Co-authored-by: Eli Bishop <[email protected]> Co-authored-by: Ben Woskow <[email protected]> Co-authored-by: LaunchDarklyCI <[email protected]> Co-authored-by: Andrew Shannon Brown <[email protected]> Co-authored-by: hroederld <[email protected]> Co-authored-by: LaunchDarklyReleaseBot <[email protected]> Co-authored-by: Dan Richelson <[email protected]> Co-authored-by: Dan Richelson <[email protected]> Co-authored-by: Ben Woskow <[email protected]> Co-authored-by: Louis Chan <[email protected]> Co-authored-by: Louis Chan <[email protected]> Co-authored-by: Moshe Good <[email protected]> Co-authored-by: Moshe Good <[email protected]>
launchdarkly · Jul 2, 2022 · 56b3e7c · 56b3e7c
1 parent caa6278
commit 56b3e7c
Show file tree

Hide file tree

Showing 7 changed files with 257 additions and 40 deletions.
diff --git a/go.mod b/go.mod
@@ -43,6 +43,7 @@ require (
 	github.com/stretchr/testify v1.7.0
 	go.opencensus.io v0.23.0
 	golang.org/x/crypto v0.0.0-20220411220226-7b82a4e95df4 // indirect
+	golang.org/x/sync v0.0.0-20220513210516-0976fa681c29 // indirect
 	golang.org/x/text v0.3.7 // indirect
 	gopkg.in/gcfg.v1 v1.2.3
 	gopkg.in/launchdarkly/go-jsonstream.v1 v1.0.1

diff --git a/go.sum b/go.sum
@@ -541,6 +541,8 @@ golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c h1:5KslGYwFpkhGh+Q16bwMP3cOontH8FOep7tGV86Y7SQ=
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20220513210516-0976fa681c29 h1:w8s32wxx3sY+OjLlv9qltkLU5yvJzxjjgiHWLjdIcw4=
+golang.org/x/sync v0.0.0-20220513210516-0976fa681c29/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=

diff --git a/internal/core/streams/stream_provider_server_side.go b/internal/core/streams/stream_provider_server_side.go
@@ -5,6 +5,7 @@ import (
 	"sync"
 
 	"github.com/launchdarkly/ld-relay/v6/config"
+	"golang.org/x/sync/singleflight"
 
 	"github.com/launchdarkly/eventsource"
 	"gopkg.in/launchdarkly/go-sdk-common.v2/ldlog"
@@ -27,6 +28,8 @@ type serverSideEnvStreamProvider struct {
 type serverSideEnvStreamRepository struct {
 	store   EnvStoreQueries
 	loggers ldlog.Loggers
+
+	flightGroup singleflight.Group
 }
 
 func (s *serverSideStreamProvider) Handler(credential config.SDKCredential) http.HandlerFunc {
@@ -92,22 +95,44 @@ func (r *serverSideEnvStreamRepository) Replay(channel, id string) chan eventsou
 	}
 	go func() {
 		defer close(out)
+		event, err := r.getReplayEvent()
+		if err != nil {
+			return
+		}
+		out <- event
+	}()
+	return out
+}
+
+// getReplayEvent will return a ServerSidePutEvent with all the data needed for a Replay.
+func (r *serverSideEnvStreamRepository) getReplayEvent() (eventsource.Event, error) {
+	data, err, _ := r.flightGroup.Do("getReplayEvent", func() (interface{}, error) {
 		flags, err := r.store.GetAll(ldstoreimpl.Features())
 
 		if err != nil {
 			r.loggers.Errorf("Error getting all flags: %s\n", err.Error())
-		} else {
-			segments, err := r.store.GetAll(ldstoreimpl.Segments())
-			if err != nil {
-				r.loggers.Errorf("Error getting all segments: %s\n", err.Error())
-			} else {
-				allData := []ldstoretypes.Collection{
-					{Kind: ldstoreimpl.Features(), Items: removeDeleted(flags)},
-					{Kind: ldstoreimpl.Segments(), Items: removeDeleted(segments)},
-				}
-				out <- MakeServerSidePutEvent(allData)
-			}
+			return nil, err
 		}
-	}()
-	return out
+		segments, err := r.store.GetAll(ldstoreimpl.Segments())
+		if err != nil {
+			r.loggers.Errorf("Error getting all segments: %s\n", err.Error())
+			return nil, err
+		}
+
+		allData := []ldstoretypes.Collection{
+			{Kind: ldstoreimpl.Features(), Items: removeDeleted(flags)},
+			{Kind: ldstoreimpl.Segments(), Items: removeDeleted(segments)},
+		}
+
+		event := MakeServerSidePutEvent(allData)
+		return event, nil
+	})
+
+	if err != nil {
+		return nil, err
+	}
+
+	// panic if it's not an eventsource.Event - as this should be impossible
+	event := data.(eventsource.Event)
+	return event, nil
 }
diff --git a/internal/core/streams/stream_provider_server_side_flags.go b/internal/core/streams/stream_provider_server_side_flags.go
@@ -5,6 +5,7 @@ import (
 	"sync"
 
 	"github.com/launchdarkly/ld-relay/v6/config"
+	"golang.org/x/sync/singleflight"
 
 	"github.com/launchdarkly/eventsource"
 	"gopkg.in/launchdarkly/go-sdk-common.v2/ldlog"
@@ -27,6 +28,8 @@ type serverSideFlagsOnlyEnvStreamProvider struct {
 type serverSideFlagsOnlyEnvStreamRepository struct {
 	store   EnvStoreQueries
 	loggers ldlog.Loggers
+
+	flightGroup singleflight.Group
 }
 
 func (s *serverSideFlagsOnlyStreamProvider) Handler(credential config.SDKCredential) http.HandlerFunc {
@@ -91,16 +94,36 @@ func (r *serverSideFlagsOnlyEnvStreamRepository) Replay(channel, id string) chan
 	}
 	go func() {
 		defer close(out)
-		if r.store.IsInitialized() {
-			flags, err := r.store.GetAll(ldstoreimpl.Features())
-
-			if err != nil {
-				r.loggers.Errorf("Error getting all flags: %s\n", err.Error())
-			} else {
-				out <- MakeServerSideFlagsOnlyPutEvent(
-					[]ldstoretypes.Collection{{Kind: ldstoreimpl.Features(), Items: removeDeleted(flags)}})
-			}
+		event, err := r.getReplayEvent()
+		if err == nil && event != nil {
+			out <- event
 		}
 	}()
 	return out
 }
+
+func (r *serverSideFlagsOnlyEnvStreamRepository) getReplayEvent() (eventsource.Event, error) {
+	data, err, _ := r.flightGroup.Do("getReplayEvent", func() (interface{}, error) {
+		if !r.store.IsInitialized() {
+			return nil, nil
+		}
+		flags, err := r.store.GetAll(ldstoreimpl.Features())
+
+		if err != nil {
+			r.loggers.Errorf("Error getting all flags: %s\n", err.Error())
+			return nil, err
+		}
+
+		event := MakeServerSideFlagsOnlyPutEvent(
+			[]ldstoretypes.Collection{{Kind: ldstoreimpl.Features(), Items: removeDeleted(flags)}})
+		return event, nil
+	})
+
+	if err != nil {
+		return nil, err
+	}
+
+	// panic if it's not an eventsource.Event - as this should be impossible
+	event := data.(eventsource.Event)
+	return event, nil
+}
diff --git a/internal/core/streams/stream_provider_server_side_flags_test.go b/internal/core/streams/stream_provider_server_side_flags_test.go
@@ -105,8 +105,10 @@ func TestStreamProviderServerSideFlagsOnly(t *testing.T) {
 	})
 
 	t.Run("initial event - store error for flags", func(t *testing.T) {
-		store := makeMockStore([]ldmodel.FeatureFlag{testFlag1, testFlag2}, []ldmodel.Segment{testSegment1})
-		store.fakeFlagsError = fakeError
+		store := newMockStoreQueries()
+		store.setupGetAllFn(func(kind ldstoretypes.DataKind) ([]ldstoretypes.KeyedItemDescriptor, error) {
+			return nil, fakeError
+		})
 
 		withStreamProvider(t, 0, func(sp StreamProvider) {
 			esp := sp.Register(validCredential, store, ldlog.NewDisabledLoggers())

diff --git a/internal/core/streams/stream_provider_server_side_test.go b/internal/core/streams/stream_provider_server_side_test.go
@@ -1,13 +1,17 @@
 package streams
 
 import (
+	"encoding/json"
+	"sync"
 	"testing"
 	"time"
 
+	"github.com/launchdarkly/eventsource"
 	"github.com/launchdarkly/ld-relay/v6/internal/basictypes"
 	"github.com/launchdarkly/ld-relay/v6/internal/core/sharedtest"
 
 	"gopkg.in/launchdarkly/go-sdk-common.v2/ldlog"
+	"gopkg.in/launchdarkly/go-server-sdk-evaluation.v1/ldbuilders"
 	"gopkg.in/launchdarkly/go-server-sdk-evaluation.v1/ldmodel"
 	"gopkg.in/launchdarkly/go-server-sdk.v5/interfaces/ldstoretypes"
 	"gopkg.in/launchdarkly/go-server-sdk.v5/ldcomponents/ldstoreimpl"
@@ -106,8 +110,13 @@ func TestStreamProviderServerSide(t *testing.T) {
 	})
 
 	t.Run("initial event - store error for flags", func(t *testing.T) {
-		store := makeMockStore([]ldmodel.FeatureFlag{testFlag1, testFlag2}, []ldmodel.Segment{testSegment1})
-		store.fakeFlagsError = fakeError
+		store := newMockStoreQueries()
+		store.setupGetAllFn(func(kind ldstoretypes.DataKind) ([]ldstoretypes.KeyedItemDescriptor, error) {
+			if kind == ldstoreimpl.Features() {
+				return nil, fakeError
+			}
+			return nil, nil
+		})
 
 		withStreamProvider(t, 0, func(sp StreamProvider) {
 			esp := sp.Register(validCredential, store, ldlog.NewDisabledLoggers())
@@ -119,8 +128,13 @@ func TestStreamProviderServerSide(t *testing.T) {
 	})
 
 	t.Run("initial event - store error for segments", func(t *testing.T) {
-		store := makeMockStore([]ldmodel.FeatureFlag{testFlag1, testFlag2}, []ldmodel.Segment{testSegment1})
-		store.fakeSegmentsError = fakeError
+		store := newMockStoreQueries()
+		store.setupGetAllFn(func(kind ldstoretypes.DataKind) ([]ldstoretypes.KeyedItemDescriptor, error) {
+			if kind == ldstoreimpl.Segments() {
+				return nil, fakeError
+			}
+			return nil, nil
+		})
 
 		withStreamProvider(t, 0, func(sp StreamProvider) {
 			esp := sp.Register(validCredential, store, ldlog.NewDisabledLoggers())
@@ -202,4 +216,107 @@ func TestStreamProviderServerSide(t *testing.T) {
 			verifyHandlerHeartbeat(t, sp, esp, validCredential)
 		})
 	})
+
+	t.Run("Replay", func(t *testing.T) {
+		const flagKey = "flagkey"
+
+		expectReplayedEvents := func(t *testing.T, eventChannel <-chan eventsource.Event) []eventsource.Event {
+			out := make([]eventsource.Event, 0)
+			for {
+				select {
+				case e, ok := <-eventChannel:
+					if !ok {
+						return out // channel was closed; this is expected after the last event
+					}
+					out = append(out, e)
+				case <-time.After(time.Second):
+					require.Fail(t, "timed out waiting for replayed event (channel was not closed)")
+				}
+			}
+		}
+
+		queryThatIncrementsFlagVersionOnEachCall := func() func(kind ldstoretypes.DataKind) ([]ldstoretypes.KeyedItemDescriptor, error) {
+			nextVersion := 1
+			return func(kind ldstoretypes.DataKind) ([]ldstoretypes.KeyedItemDescriptor, error) {
+				if kind != ldstoreimpl.Features() {
+					return nil, nil
+				}
+				flag := ldbuilders.NewFlagBuilder("flagkey").Version(nextVersion).Build()
+				nextVersion++
+				return []ldstoretypes.KeyedItemDescriptor{
+					{Key: flag.Key, Item: sharedtest.FlagDesc(flag)},
+				}, nil
+			}
+		}
+
+		getFlagFromEventData := func(t *testing.T, e eventsource.Event) ldmodel.FeatureFlag {
+			require.Equal(t, "put", e.Event())
+			var data struct {
+				Data struct {
+					Flags map[string]ldmodel.FeatureFlag `json:"flags"`
+				} `json:"data"`
+			}
+			require.NoError(t, json.Unmarshal([]byte(e.Data()), &data))
+			require.Contains(t, data.Data.Flags, flagKey)
+			return data.Data.Flags[flagKey]
+		}
+
+		t.Run("second client connects after first computation is done", func(t *testing.T) {
+			store := newMockStoreQueries()
+			store.setupGetAllFn(queryThatIncrementsFlagVersionOnEachCall())
+			repo := &serverSideEnvStreamRepository{store: store, loggers: ldlog.NewDisabledLoggers()}
+
+			eventCh1 := repo.Replay("", "")
+			events1 := expectReplayedEvents(t, eventCh1)
+			require.Len(t, events1, 1)
+
+			eventCh2 := repo.Replay("", "")
+			events2 := expectReplayedEvents(t, eventCh2)
+			require.Len(t, events2, 1)
+
+			assert.Equal(t, 1, getFlagFromEventData(t, events1[0]).Version)
+			assert.Equal(t, 2, getFlagFromEventData(t, events2[0]).Version) // two separate computations were done
+		})
+
+		t.Run("second client connects while first computation is still in progress", func(t *testing.T) {
+			underlyingQuery := queryThatIncrementsFlagVersionOnEachCall()
+			replayStarted := make(chan struct{}, 2)
+			replayCanFinish := make(chan struct{}, 1)
+			var gateFirstReplay sync.Once
+			store := newMockStoreQueries()
+			store.setupGetAllFn(func(kind ldstoretypes.DataKind) ([]ldstoretypes.KeyedItemDescriptor, error) {
+				if kind != ldstoreimpl.Features() {
+					return nil, nil
+				}
+				replayStarted <- struct{}{}
+				ret, err := underlyingQuery(kind)
+				gateFirstReplay.Do(func() {
+					<-replayCanFinish
+				})
+
+				time.Sleep(time.Millisecond * 200)
+				// This delay is arbitrary and possibly overly timing-sensitive, but it looks like there is no
+				// way to really guarantee that the goroutine for the second Replay has started before we allow
+				// the first one to complete, without adding just-for-tests instrumentation inside of
+				// serverSideEnvStreamRepository.getReplayEvent().
+
+				return ret, err
+			})
+			repo := &serverSideEnvStreamRepository{store: store, loggers: ldlog.NewDisabledLoggers()}
+
+			eventCh1 := repo.Replay("", "")
+			<-replayStarted
+			eventCh2 := repo.Replay("", "")
+			replayCanFinish <- struct{}{}
+
+			events1 := expectReplayedEvents(t, eventCh1)
+			require.Len(t, events1, 1)
+
+			events2 := expectReplayedEvents(t, eventCh2)
+			require.Len(t, events2, 1)
+
+			assert.Equal(t, 1, getFlagFromEventData(t, events1[0]).Version)
+			assert.Equal(t, 1, getFlagFromEventData(t, events2[0]).Version) // only one computation was done
+		})
+	})
 }