[Choreo] Upstream connection handling support considering TCP and HTTP

adapter/cmd/adapter/main.go

@@ -42,6 +42,7 @@ func startMicroGateway() {
 		logger.Fatal("Error starting the adapter", err)
 	}
 	conf, errReadConfig := config.ReadConfigs()
+	config.GetTCPKeepaliveEnabledOrgs()
 	if errReadConfig != nil {
 		logger.Fatal("Error loading configuration. ", errReadConfig)
 	}

adapter/config/default_config.go

@@ -157,6 +157,14 @@ var defaultConfig = &Config{
 					MaxRetries:         50,
 				},
 			},
+			TCPConfigurations: upstreamTCPConfigs{
+				KeepaliveTimeInMillis:     120000,
+				KeepaliveProbes:           9,
+				KeepaliveIntervalInMillis: 75000,
+			},
+			HTTPConfigurations: upstreamHTTPConfigs{
+				IdleTimeoutInMillis:           120000,
+			},
 		},
 		Connection: connection{
 			Timeouts: connectionTimeouts{

adapter/config/parser.go

@@ -34,11 +34,13 @@ import (
 )
 
 var (
-	onceConfigRead      sync.Once
-	onceGetDefaultVhost sync.Once
-	adapterConfig       *Config
-	defaultVhost        map[string]string
-	e                   error
+	onceConfigRead                       sync.Once
+	onceGetDefaultVhost                  sync.Once
+	adapterConfig                        *Config
+	defaultVhost                         map[string]string
+	e                                    error
+	// UpstreamConnectionConfEnabledOrgList is the list of orgs that need to handle connection timeouts
+	UpstreamConnectionConfEnabledOrgList []string
 )
 
 // DefaultGatewayName represents the name of the default gateway
@@ -255,3 +257,12 @@ func (config *Config) validateConfig() error {
 func printDeprecatedWarningLog(deprecatedTerm, currentTerm string) {
 	logger.Warnf("%s is deprecated. Use %s instead", deprecatedTerm, currentTerm)
 }
+
+// GetTCPKeepaliveEnabledOrgs returns the list of orgs that need to handle connection timeouts
+func GetTCPKeepaliveEnabledOrgs() {
+	orgs := os.Getenv("TCP_KEEPALIVE_ENABLED_ORGS")
+	UpstreamConnectionConfEnabledOrgList = strings.Split(orgs, ",")
+	if len(UpstreamConnectionConfEnabledOrgList) == 0 {
+		UpstreamConnectionConfEnabledOrgList[0] = ""
+	}
+}

adapter/config/types.go

@@ -235,12 +235,14 @@ type globalCors struct {
 // Envoy Upstream Related Configurations
 type envoyUpstream struct {
 	// UpstreamTLS related Configuration
-	TLS             upstreamTLS
-	Timeouts        upstreamTimeout
-	Health          upstreamHealth
-	DNS             upstreamDNS
-	Retry           upstreamRetry
-	CircuitBreakers []upstreamCircuitBreaker
+	TLS                upstreamTLS
+	Timeouts           upstreamTimeout
+	Health             upstreamHealth
+	DNS                upstreamDNS
+	Retry              upstreamRetry
+	CircuitBreakers    []upstreamCircuitBreaker
+	TCPConfigurations  upstreamTCPConfigs
+	HTTPConfigurations upstreamHTTPConfigs
 }
 
 type upstreamTLS struct {
@@ -276,6 +278,15 @@ type dnsResolverConfig struct {
 	CAres        cAres
 }
 
+type upstreamTCPConfigs struct {
+	KeepaliveTimeInMillis     uint32
+	KeepaliveProbes           uint32
+	KeepaliveIntervalInMillis uint32
+}
+type upstreamHTTPConfigs struct {
+	IdleTimeoutInMillis           uint32
+}
+
 type dnsResolverType string
 
 const (

adapter/internal/api/apis_impl.go

@@ -166,7 +166,11 @@ func ProcessMountedAPIProjects() (err error) {
 
 func validateAndUpdateXds(apiProject mgw.ProjectAPI, override *bool) (err error) {
 	apiYaml := apiProject.APIYaml.Data
-	apiProject.OrganizationID = config.GetControlPlaneConnectedTenantDomain()
+	if (apiProject.APIYaml.Data.OrganizationID != "") {
+		apiProject.OrganizationID = apiProject.APIYaml.Data.OrganizationID
+	} else {
+		apiProject.OrganizationID = config.GetControlPlaneConnectedTenantDomain()
+	}
 
 	// handle panic
 	defer func() {

adapter/internal/oasparser/envoyconf/routes_with_clusters.go

@@ -129,6 +129,11 @@ func CreateRoutesWithClusters(mgwSwagger model.MgwSwagger, upstreamCerts map[str
 		if !strings.Contains(apiLevelEndpointProd.EndpointPrefix, xWso2EPClustersConfigNamePrefix) {
 			cluster, address, err := processEndpoints(apiLevelClusterNameProd, apiLevelEndpointProd,
 				upstreamCerts, timeout, apiLevelbasePath)
+			// assigns specified values for TCP keep-alive and HTTP timeout considering specific organizations
+			ok := slices.Contains(config.UpstreamConnectionConfEnabledOrgList, organizationID) || slices.Contains(config.UpstreamConnectionConfEnabledOrgList, "*")
+			if ok {
+				getKeepAliveConfigs(cluster, conf)
+			}
 			if err != nil {
 				apiLevelClusterNameProd = ""
 				logger.LoggerOasparser.Errorf("Error while adding api level production endpoints for %s. %v , skipping api...", apiTitle, err.Error())
@@ -343,6 +348,33 @@ func getClusterName(epPrefix string, organizationID string, vHost string, swagge
 		swaggerVersion)
 }
 
+func getKeepAliveConfigs(cluster *clusterv3.Cluster, conf *config.Config) {
+	cluster.UpstreamConnectionOptions = &clusterv3.UpstreamConnectionOptions{
+		TcpKeepalive: &corev3.TcpKeepalive{
+			KeepaliveProbes:   wrapperspb.UInt32(conf.Envoy.Upstream.TCPConfigurations.KeepaliveProbes),
+			KeepaliveInterval: wrapperspb.UInt32(conf.Envoy.Upstream.TCPConfigurations.KeepaliveIntervalInMillis / 1000),
+			KeepaliveTime:     wrapperspb.UInt32(conf.Envoy.Upstream.TCPConfigurations.KeepaliveTimeInMillis / 1000),
+		},
+	}
+
+	config := &upstreams.HttpProtocolOptions{
+		CommonHttpProtocolOptions: &corev3.HttpProtocolOptions{
+			IdleTimeout: durationpb.New(time.Duration(conf.Envoy.Upstream.HTTPConfigurations.IdleTimeoutInMillis) * time.Millisecond),
+		},
+		UpstreamProtocolOptions: &upstreams.HttpProtocolOptions_UseDownstreamProtocolConfig{},
+	}
+	MarshalledHTTPProtocolOptions, err := proto.Marshal(config)
+	if err != nil {
+		logger.LoggerOasparser.Error("Error while marshalling the upstream TCP keep alive config")
+	}
+	cluster.TypedExtensionProtocolOptions = map[string]*anypb.Any{
+		"envoy.extensions.upstreams.http.v3.HttpProtocolOptions": {
+			TypeUrl: httpProtocolOptionsName,
+			Value:   MarshalledHTTPProtocolOptions,
+		},
+	}
+}
+
 // CreateLuaCluster creates lua cluster configuration.
 func CreateLuaCluster(interceptorCerts map[string][]byte, endpoint model.InterceptEndpoint) (*clusterv3.Cluster, []*corev3.Address, error) {
 	logger.LoggerOasparser.Debug("creating a lua cluster ", endpoint.ClusterName)

resources/conf/config.toml.template

@@ -189,6 +189,20 @@ retainKeys = ["self_validate_jwt", "issuer", "claim_mappings", "consumer_key_cla
   # Max interval for the Envoy's exponential retry back off algorithm
   maxInterval = "500ms"
 
+# TCP configurations applicable with the upstream clusters
+[router.upstream.tCPConfigurations]
+  # The number of milliseconds a connection needs to be idle before keep-alive probes start being sent
+  keepaliveTimeInMillis = 120000
+  # Maximum number of keep-alive probes to send without response before deciding the connection is dead
+  keepaliveProbes = 9
+  # The number of milliseconds between keep-alive probes
+  keepaliveIntervalInMillis = 75000
+
+# HTTP configurations applicable with the upstream clusters
+[router.upstream.hTTPConfigurations]
+  # Idle timeout in milliseconds for connections
+  idleTimeoutInMillis = 120000
+
 # Timeouts managed by the connection manager
 [router.connectionTimeout]
   # The amount of time that Envoy will wait for the entire request to be received. Time from client to upstream.