[7.x] Update kibana endpoint and introduce a killer switch to disable the (#2386)

agentcfg/fetch.go

@@ -33,7 +33,7 @@ import (
 )
 
 const (
-	endpoint = "/api/apm/settings/cm/search"
+	endpoint = "/api/apm/settings/agent-configuration/search"
 )
 
 // Fetcher holds static information and information shared between requests.

beater/agent_config_handler.go

@@ -37,7 +37,7 @@ const (
 	errMaxAgeDuration  = 5 * time.Minute
 )
 
-func agentConfigHandler(kbClient *kibana.Client, config *agentConfig, secretToken string) http.Handler {
+func agentConfigHandler(kbClient *kibana.Client, enabled bool, config *agentConfig, secretToken string) http.Handler {
 	fetcher := agentcfg.NewFetcher(kbClient, config.Cache.Expiration)
 	defaultHeaderCacheControl := fmt.Sprintf("max-age=%v, must-revalidate", config.Cache.Expiration.Seconds())
 	errHeaderCacheControl := fmt.Sprintf("max-age=%v, must-revalidate", errMaxAgeDuration.Seconds())
@@ -64,7 +64,7 @@ func agentConfigHandler(kbClient *kibana.Client, config *agentConfig, secretToke
 			headerCacheControlVal = errHeaderCacheControl
 		case internalErr != nil:
 			resp = internalErr.Error()
-			state = http.StatusInternalServerError
+			state = http.StatusServiceUnavailable
 			headerCacheControlVal = errHeaderCacheControl
 		case len(cfg) == 0:
 			resp = nil
@@ -91,7 +91,9 @@ func agentConfigHandler(kbClient *kibana.Client, config *agentConfig, secretToke
 				"error", resp)
 		}
 	})
-	return logHandler(authHandler(secretToken, handler))
+	return logHandler(
+		killSwitchHandler(enabled,
+			authHandler(secretToken, handler)))
 }
 
 // Returns (zero, error) if request body can't be unmarshalled or service.name is missing

beater/agent_config_handler_test.go

@@ -98,7 +98,7 @@ var testcases = map[string]struct {
 		),
 		method:                 http.MethodGet,
 		queryParams:            map[string]string{"service.name": "opbeans-ruby"},
-		respStatus:             http.StatusInternalServerError,
+		respStatus:             http.StatusServiceUnavailable,
 		respCacheControlHeader: "max-age=300, must-revalidate",
 		respBody:               true,
 	},
@@ -132,7 +132,7 @@ func TestAgentConfigHandler(t *testing.T) {
 
 	for name, tc := range testcases {
 		t.Run(name, func(t *testing.T) {
-			h := agentConfigHandler(tc.kbClient, &cfg, "")
+			h := agentConfigHandler(tc.kbClient, true, &cfg, "")
 			w := httptest.NewRecorder()
 			r := httptest.NewRequest(tc.method, target(tc.queryParams), nil)
 			for k, v := range tc.requestHeader {
@@ -152,6 +152,17 @@ func TestAgentConfigHandler(t *testing.T) {
 	}
 }
 
+func TestAgentConfigDisabled(t *testing.T) {
+	cfg := agentConfig{Cache: &Cache{Expiration: time.Nanosecond}}
+	h := agentConfigHandler(nil, false, &cfg, "")
+
+	w := httptest.NewRecorder()
+	r := httptest.NewRequest(http.MethodGet, "/config", nil)
+	h.ServeHTTP(w, r)
+
+	assert.Equal(t, http.StatusForbidden, w.Code, w.Body.String())
+}
+
 func TestAgentConfigHandlerPostOk(t *testing.T) {
 
 	kb := tests.MockKibana(http.StatusOK, m{
@@ -164,7 +175,7 @@ func TestAgentConfigHandlerPostOk(t *testing.T) {
 	})
 
 	var cfg = agentConfig{Cache: &Cache{Expiration: time.Nanosecond}}
-	h := agentConfigHandler(kb, &cfg, "")
+	h := agentConfigHandler(kb, true, &cfg, "")
 
 	w := httptest.NewRecorder()
 	r := httptest.NewRequest(http.MethodPost, "/config", convert.ToReader(m{

beater/common_handler_test.go

@@ -34,13 +34,16 @@ func TestIncCounter(t *testing.T) {
 	req.Header.Set("Accept", "application/json")
 	w := httptest.NewRecorder()
 
-	for i := 1; i <= 5; i++ {
-		for _, res := range []serverResponse{acceptedResponse, okResponse, forbiddenResponse(errors.New("")), unauthorizedResponse,
-			requestTooLargeResponse, rateLimitedResponse, methodNotAllowedResponse,
-			cannotValidateResponse(errors.New("")), cannotDecodeResponse(errors.New("")),
-			fullQueueResponse(errors.New("")), serverShuttingDownResponse(errors.New(""))} {
+	responseCounter.Set(0)
+	responseErrors.Set(0)
+	for _, res := range []serverResponse{acceptedResponse, okResponse, forbiddenResponse(errors.New("")), unauthorizedResponse,
+		requestTooLargeResponse, rateLimitedResponse, methodNotAllowedResponse,
+		cannotValidateResponse(errors.New("")), cannotDecodeResponse(errors.New("")),
+		fullQueueResponse(errors.New("")), serverShuttingDownResponse(errors.New(""))} {
+		res.counter.Set(0)
+		for i := 1; i <= 5; i++ {
 			sendStatus(w, req, res)
-			assert.Equal(t, int64(i), res.counter.Get())
+			assert.Equal(t, int64(i), res.counter.Get(), string(res.code))
 		}
 	}
 	assert.Equal(t, int64(55), responseCounter.Get())

beater/route_config.go

@@ -114,7 +114,7 @@ func newMuxer(beaterConfig *Config, kbClient *kibana.Client, report publish.Repo
 		mux.Handle(path, handler)
 	}
 
-	mux.Handle(agentConfigURL, agentConfigHandler(kbClient, beaterConfig.AgentConfig, beaterConfig.SecretToken))
+	mux.Handle(agentConfigURL, agentConfigHandler(kbClient, beaterConfig.Kibana.Enabled(), beaterConfig.AgentConfig, beaterConfig.SecretToken))
 	logger.Infof("Path %s added to request handler", agentConfigURL)
 
 	mux.Handle(rootURL, rootHandler(beaterConfig.SecretToken))

tests/system/config/apm-server.yml.j2

@@ -77,6 +77,11 @@ apm-server:
   ilm.enabled: {{ ilm_enabled }}
   {% endif %}
 
+  {% if kibana_enabled %}
+  kibana.enabled: {{ kibana_enabled }}
+  {% endif %}
+
+
 ############################# Setup ##########################################
 
 {% if override_template %}

tests/system/test_integration_logging.py

@@ -53,6 +53,7 @@ class LoggingIntegrationAuth(ElasticTest):
     config_overrides = {
         "logging_json": "true",
         "secret_token": "supersecret",
+        "kibana_enabled": "true",
     }
 
     @unittest.skipUnless(INTEGRATION_TESTS, "integration test")
@@ -69,7 +70,7 @@ def test_log_config_request(self):
                               "Content-Type": "application/x-ndjson",
                               "Authorization": "Bearer " + self.config_overrides["secret_token"],
                           })
-        assert r2.status_code == 500, r2.status_code
+        assert r2.status_code == 503, r2.status_code
 
         config_request_logs = list(self.logged_requests(url="/config/v1/agents"))
         assert len(config_request_logs) == 2, config_request_logs
@@ -82,11 +83,31 @@ def test_log_config_request(self):
         self.assertDictContainsSubset({
             "level": "error",
             "message": "error handling request",
-            "error": "no configured Kibana Client: provide apm-server.kibana.* settings",
-            "response_code": 500,
+            "response_code": 503,
         }, config_request_logs[1])
 
 
+class LoggingIntegrationAuth2(ElasticTest):
+    config_overrides = {
+        "logging_json": "true",
+    }
+
+    @unittest.skipUnless(INTEGRATION_TESTS, "integration test")
+    def test_log_kill_switch_active(self):
+        r = requests.get(self.agent_config_url,
+                         headers={
+                             "Content-Type": "application/x-ndjson",
+                         })
+        assert r.status_code == 403, r.status_code
+        config_request_logs = list(self.logged_requests(url="/config/v1/agents"))
+        self.assertDictContainsSubset({
+            "level": "error",
+            "message": "error handling request",
+            "error": {"error": "forbidden request: endpoint is disabled"},
+            "response_code": 403,
+        }, config_request_logs[0])
+
+
 class LoggingIntegrationEventSizeTest(ElasticTest):
     config_overrides = {
         "logging_json": "true",