golang: add DrainConnectionUponCompletion() to StreamInfo API (envoyproxy#43037)

Commit Message:
This change exposes the `setShouldDrainConnectionUponCompletion`
functionality to the Go HTTP filter, allowing Go plugins to mark
connections for draining after the current request completes.

For HTTP/1.x, this will add a "Connection: close" header and close the
connection after the response is sent. For HTTP/2 and HTTP/3, this will
send a GOAWAY frame and initiate a graceful drain sequence.

This mirrors the functionality added to the Lua filter in 1.37
(drainConnectionUponCompletion).

Additional Description:
Risk Level:
Testing:
Docs Changes:
Release Notes:
Platform Specific Features:
[Optional Runtime guard:]
[Optional Fixes #Issue]
[Optional Fixes commit #PR or SHA]
[Optional Deprecated:]
[Optional [API
Considerations](https://github.com/envoyproxy/envoy/blob/main/api/review_checklist.md):]

Signed-off-by: William Dauchy <william.dauchy@datadoghq.com>

Author: wdauchy

contrib/golang/common/go/api/api.h

@@ -123,6 +123,7 @@ CAPIStatus envoyGoFilterHttpGetStringProperty(void* r, void* key_data, int key_l
                                               uint64_t* value_data, int* value_len, int* rc);
 CAPIStatus envoyGoFilterHttpGetStringSecret(void* r, void* key_data, int key_len,
                                             uint64_t* value_data, int* value_len);
+CAPIStatus envoyGoFilterHttpSetDrainConnectionUponCompletion(void* r);
 
 /* These APIs have nothing to do with request */
 void envoyGoFilterLog(uint32_t level, void* message_data, int message_len);

contrib/golang/common/go/api/capi.go

@@ -61,6 +61,7 @@ type HttpCAPI interface {
 
 	HttpFinalize(r unsafe.Pointer, reason int)
 	HttpGetStringSecret(c unsafe.Pointer, key string) (string, bool)
+	HttpSetDrainConnectionUponCompletion(r unsafe.Pointer)
 
 	/* These APIs are related to config, use the pointer of config. */
 	HttpDefineMetric(c unsafe.Pointer, metricType MetricType, name string) uint32

contrib/golang/common/go/api/filter.go

@@ -160,6 +160,10 @@ type StreamInfo interface {
 	VirtualClusterName() (string, bool)
 	// WorkerID returns the ID of the Envoy worker thread
 	WorkerID() uint32
+	// DrainConnectionUponCompletion marks the connection to be drained after the current request completes.
+	// For HTTP/1.x, this will add a "Connection: close" header to the response.
+	// For HTTP/2 and HTTP/3, this will send a GOAWAY frame after the response is sent.
+	DrainConnectionUponCompletion()
 	// Some fields in stream info can be fetched via GetProperty
 	// For example, startTime() is equal to GetProperty("request.time")
 }

contrib/golang/filters/http/source/cgo.cc

@@ -370,6 +370,12 @@ CAPIStatus envoyGoFilterHttpGetStringSecret(void* r, void* key_data, int key_len
       });
 }
 
+CAPIStatus envoyGoFilterHttpSetDrainConnectionUponCompletion(void* r) {
+  return envoyGoFilterHandlerWrapper(r, [](std::shared_ptr<Filter>& filter) -> CAPIStatus {
+    return filter->setDrainConnectionUponCompletion();
+  });
+}
+
 CAPIStatus envoyGoFilterHttpDefineMetric(void* c, uint32_t metric_type, void* name_data,
                                          int name_len, uint32_t* metric_id) {
   return envoyGoConfigHandlerWrapper(

contrib/golang/filters/http/source/go/pkg/http/capi_impl.go

@@ -490,6 +490,12 @@ func (c *httpCApiImpl) HttpGetStringSecret(r unsafe.Pointer, key string) (string
 	return strings.Clone(unsafe.String((*byte)(unsafe.Pointer(uintptr(valueData))), int(valueLen))), true
 }
 
+func (c *httpCApiImpl) HttpSetDrainConnectionUponCompletion(r unsafe.Pointer) {
+	req := (*httpRequest)(r)
+	res := C.envoyGoFilterHttpSetDrainConnectionUponCompletion(unsafe.Pointer(req.req))
+	handleCApiStatus(res)
+}
+
 func (c *httpCApiImpl) HttpLog(level api.LogType, message string) {
 	C.envoyGoFilterLog(C.uint32_t(level), unsafe.Pointer(unsafe.StringData(message)), C.int(len(message)))
 }

contrib/golang/filters/http/source/go/pkg/http/filter.go

@@ -391,6 +391,10 @@ func (s *streamInfo) WorkerID() uint32 {
 	return uint32(s.request.req.worker_id)
 }
 
+func (s *streamInfo) DrainConnectionUponCompletion() {
+	cAPI.HttpSetDrainConnectionUponCompletion(unsafe.Pointer(s.request))
+}
+
 type filterState struct {
 	request *httpRequest
 }

contrib/golang/filters/http/source/golang_filter.cc

@@ -1546,6 +1546,16 @@ CAPIStatus Filter::getSecret(const absl::string_view name, uint64_t* value_data,
   }
 }
 
+CAPIStatus Filter::setDrainConnectionUponCompletion() {
+  Thread::LockGuard lock(mutex_);
+  if (has_destroyed_) {
+    ENVOY_LOG(debug, "golang filter has been destroyed");
+    return CAPIStatus::CAPIFilterIsDestroy;
+  }
+  streamInfo().setShouldDrainConnectionUponCompletion(true);
+  return CAPIStatus::CAPIOK;
+}
+
 /* ConfigId */
 
 uint64_t Filter::getMergedConfigId() {

contrib/golang/filters/http/source/golang_filter.h

@@ -313,6 +313,7 @@ class Filter : public Http::StreamFilter,
   CAPIStatus getStringProperty(absl::string_view path, uint64_t* value_data, int* value_len,
                                GoInt32* rc);
   CAPIStatus getSecret(absl::string_view key, uint64_t* value_data, int* value_len);
+  CAPIStatus setDrainConnectionUponCompletion();
 
   bool isProcessingInGo() {
     return decoding_state_.isProcessingInGo() || encoding_state_.isProcessingInGo();

contrib/golang/filters/http/test/golang_integration_test.cc

@@ -1940,4 +1940,28 @@ TEST_P(GolangIntegrationTest, SetUpstreamOverrideHost_InvalidHost_Strict_NotFoun
       false, "", true);
 }
 
+// Test DrainConnectionUponCompletion triggers connection draining for HTTP/1.1.
+TEST_P(GolangIntegrationTest, DrainConnectionUponCompletion) {
+  initializeBasicFilter(BASIC);
+  registerTestServerPorts({"http"});
+
+  codec_client_ = makeHttpConnection(lookupPort("http"));
+
+  // Make request with drainConnection query parameter.
+  Http::TestRequestHeaderMapImpl request_headers{
+      {":method", "GET"}, {":path", "/test?drainConnection=1"}, {":authority", "test"}};
+  auto response = sendRequestAndWaitForResponse(request_headers, 0, default_response_headers_, 0);
+
+  EXPECT_TRUE(response->complete());
+  EXPECT_EQ("200", response->headers().getStatusValue());
+
+  // For HTTP/1.1, we should see Connection: close header.
+  EXPECT_EQ("close", response->headers().getConnectionValue());
+
+  // Connection should be closed after request completes.
+  ASSERT_TRUE(codec_client_->waitForDisconnect());
+
+  cleanupUpstreamAndDownstream();
+}
+
 } // namespace Envoy

contrib/golang/filters/http/test/test_data/basic/filter.go

@@ -41,6 +41,7 @@ type filter struct {
 
 	upstreamOverrideHost       string // set upstream override host
 	upstreamOverrideHostStrict bool   // set strict mode for upstream override host
+	drainConnection            bool   // drain connection upon completion
 }
 
 func parseQuery(path string) url.Values {
@@ -90,6 +91,7 @@ func (f *filter) initRequest(header api.RequestHeaderMap) {
 	f.refreshRoute = f.query_params.Get("refreshRoute") != ""
 	f.upstreamOverrideHost = f.query_params.Get("upstreamOverrideHost")
 	f.upstreamOverrideHostStrict = f.query_params.Get("upstreamOverrideHostStrict") != ""
+	f.drainConnection = f.query_params.Get("drainConnection") != ""
 }
 
 func (f *filter) fail(callbacks api.FilterProcessCallbacks, msg string, a ...any) api.StatusType {
@@ -285,6 +287,9 @@ func (f *filter) decodeHeaders(header api.RequestHeaderMap, endStream bool) api.
 		f.callbacks.RefreshRouteCache()
 		header.SetPath("/api/") // path used by the upstream
 	}
+	if f.drainConnection {
+		f.callbacks.StreamInfo().DrainConnectionUponCompletion()
+	}
 	return api.Continue
 }