Hardcode troughput settings for cloud environment (#5402)

cmd/root.go

@@ -19,6 +19,7 @@ package cmd
 
 import (
 	"fmt"
+	"os"
 
 	"github.com/spf13/pflag"
 
@@ -37,22 +38,68 @@ import (
 const (
 	beatName        = "apm-server"
 	apmIndexPattern = "apm"
+	cloudEnv        = "CLOUD_APM_CAPACITY"
 )
 
-var libbeatConfigOverrides = []cfgfile.ConditionalOverride{{
-	Check: func(_ *common.Config) bool {
-		return true
+type throughputSettings struct {
+	worker      int
+	bulkMaxSize int
+	events      int
+	minEvents   int
+}
+
+var cloudMatrix = map[string]throughputSettings{
+	"512":  {5, 267, 2000, 267},
+	"1024": {7, 381, 4000, 381},
+	"2048": {10, 533, 8000, 533},
+	"4096": {14, 762, 16000, 762},
+	"8192": {20, 1067, 32000, 1067},
+}
+
+var libbeatConfigOverrides = func() []cfgfile.ConditionalOverride {
+	return []cfgfile.ConditionalOverride{{
+		Check: func(_ *common.Config) bool {
+			return true
+		},
+		Config: common.MustNewConfigFrom(map[string]interface{}{
+			"logging": map[string]interface{}{
+				"metrics": map[string]interface{}{
+					"enabled": false,
+				},
+				"ecs":  true,
+				"json": true,
+			},
+		}),
 	},
-	Config: common.MustNewConfigFrom(map[string]interface{}{
-		"logging": map[string]interface{}{
-			"metrics": map[string]interface{}{
-				"enabled": false,
+		{
+			Check: func(_ *common.Config) bool {
+				return true
 			},
-			"ecs":  true,
-			"json": true,
+			Config: func() *common.Config {
+				cap := os.Getenv(cloudEnv)
+				if _, ok := cloudMatrix[cap]; !ok {
+					return common.NewConfig()
+				}
+				return common.MustNewConfigFrom(map[string]interface{}{
+					"output": map[string]interface{}{
+						"elasticsearch": map[string]interface{}{
+							"worker":        cloudMatrix[cap].worker,
+							"bulk_max_size": cloudMatrix[cap].bulkMaxSize,
+						},
+					},
+					"queue": map[string]interface{}{
+						"mem": map[string]interface{}{
+							"events": cloudMatrix[cap].events,
+							"flush": map[string]interface{}{
+								"min_events": cloudMatrix[cap].minEvents,
+							},
+						},
+					},
+				})
+			}(),
 		},
-	}),
-}}
+	}
+}
 
 // DefaultSettings return the default settings for APM Server to pass into
 // the GenRootCmdWithSettings.
@@ -67,7 +114,7 @@ func DefaultSettings() instance.Settings {
 		},
 		IndexManagement: idxmgmt.MakeDefaultSupporter,
 		Processing:      processing.MakeDefaultObserverSupport(false),
-		ConfigOverrides: libbeatConfigOverrides,
+		ConfigOverrides: libbeatConfigOverrides(),
 	}
 }
 

cmd/root_test.go

@@ -0,0 +1,53 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you 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 cmd
+
+import (
+	"os"
+	"testing"
+
+	"github.com/elastic/beats/v7/libbeat/common"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestCloudEnv(t *testing.T) {
+	defer os.Unsetenv(cloudEnv)
+
+	// no cloud environment variable set
+	settings := DefaultSettings()
+	assert.Len(t, settings.ConfigOverrides, 2)
+	assert.Equal(t, common.NewConfig(), settings.ConfigOverrides[1].Config)
+
+	// cloud environment picked up
+	os.Setenv(cloudEnv, "512")
+	settings = DefaultSettings()
+	assert.Len(t, settings.ConfigOverrides, 2)
+	cfg := settings.ConfigOverrides[1].Config
+	assert.NotNil(t, cfg)
+	workers, err := cfg.Int("output.elasticsearch.worker", -1)
+	require.NoError(t, err)
+	assert.Equal(t, int64(5), workers)
+
+	// bad cloud environment value
+	os.Setenv(cloudEnv, "123")
+	settings = DefaultSettings()
+	assert.Len(t, settings.ConfigOverrides, 2)
+	assert.Equal(t, common.NewConfig(), settings.ConfigOverrides[1].Config)
+}