Add nginx plus api actions

api/grpc/mpi/v1/command.proto

@@ -204,8 +204,11 @@ message APIActionRequest {
 message NGINXPlusAction {
     // types of actions possible with NGINX Plus API
     oneof action {
-        UpdateHTTPUpstreamServers update_http_upstream_servers = 2;
-        GetHTTPUpstreamServers get_http_upstream_servers = 3;
+        UpdateHTTPUpstreamServers update_http_upstream_servers = 1;
+        GetHTTPUpstreamServers get_http_upstream_servers = 2;
+        UpdateStreamServers update_stream_servers = 3;
+        GetUpstreams get_upstreams = 4;
+        GetStreamUpstreams get_stream_upstreams = 5;
     }
 }
 
@@ -223,6 +226,22 @@ message GetHTTPUpstreamServers {
     string http_upstream_name = 1;
 }
 
+// Update Upstream Stream Servers for an instance 
+message UpdateStreamServers {
+    // the name of the upstream stream
+    string  upstream_stream_name = 1;
+    // a list of upstream stream servers
+    repeated google.protobuf.Struct servers = 2;
+}
+
+// Get Upstreams for an instance 
+message GetUpstreams {
+}
+
+// Get Stream Upstream Servers for an instance 
+message GetStreamUpstreams {
+}
+
 // Request an update on a particular command
 message CommandStatusRequest {}
 

docs/proto/protos.md

@@ -52,6 +52,8 @@
     - [DataPlaneResponse](#mpi-v1-DataPlaneResponse)
     - [FileServer](#mpi-v1-FileServer)
     - [GetHTTPUpstreamServers](#mpi-v1-GetHTTPUpstreamServers)
+    - [GetStreamUpstreams](#mpi-v1-GetStreamUpstreams)
+    - [GetUpstreams](#mpi-v1-GetUpstreams)
     - [HealthRequest](#mpi-v1-HealthRequest)
     - [HostInfo](#mpi-v1-HostInfo)
     - [Instance](#mpi-v1-Instance)
@@ -74,6 +76,7 @@
     - [UpdateDataPlaneStatusRequest](#mpi-v1-UpdateDataPlaneStatusRequest)
     - [UpdateDataPlaneStatusResponse](#mpi-v1-UpdateDataPlaneStatusResponse)
     - [UpdateHTTPUpstreamServers](#mpi-v1-UpdateHTTPUpstreamServers)
+    - [UpdateStreamServers](#mpi-v1-UpdateStreamServers)
 
     - [InstanceHealth.InstanceHealthStatus](#mpi-v1-InstanceHealth-InstanceHealthStatus)
     - [InstanceMeta.InstanceType](#mpi-v1-InstanceMeta-InstanceType)
@@ -797,6 +800,26 @@ Get HTTP Upstream Servers for an instance
 
 
 
+<a name="mpi-v1-GetStreamUpstreams"></a>
+
+### GetStreamUpstreams
+Get Stream Upstream Servers for an instance
+
+
+
+
+
+
+<a name="mpi-v1-GetUpstreams"></a>
+
+### GetUpstreams
+Get Upstreams for an instance
+
+
+
+
+
+
 <a name="mpi-v1-HealthRequest"></a>
 
 ### HealthRequest
@@ -977,6 +1000,9 @@ Perform an action using the NGINX Plus API on an instance
 | ----- | ---- | ----- | ----------- |
 | update_http_upstream_servers | [UpdateHTTPUpstreamServers](#mpi-v1-UpdateHTTPUpstreamServers) |  |  |
 | get_http_upstream_servers | [GetHTTPUpstreamServers](#mpi-v1-GetHTTPUpstreamServers) |  |  |
+| update_stream_servers | [UpdateStreamServers](#mpi-v1-UpdateStreamServers) |  |  |
+| get_upstreams | [GetUpstreams](#mpi-v1-GetUpstreams) |  |  |
+| get_stream_upstreams | [GetStreamUpstreams](#mpi-v1-GetStreamUpstreams) |  |  |
 
 
 
@@ -1136,6 +1162,22 @@ Update HTTP Upstream Servers for an instance
 
 
 
+
+<a name="mpi-v1-UpdateStreamServers"></a>
+
+### UpdateStreamServers
+Update Upstream Stream Servers for an instance
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| upstream_stream_name | [string](#string) |  | the name of the upstream stream |
+| servers | [google.protobuf.Struct](#google-protobuf-Struct) | repeated | a list of upstream stream servers |
+
+
+
+
+
 
 
 

internal/resource/resource_plugin.go

@@ -44,6 +44,8 @@ type plusAPIErr struct {
 	Href      string      `json:"href"`
 }
 
+const emptyResponse = "{}"
+
 var _ bus.Plugin = (*Resource)(nil)
 
 func NewResource(agentConfig *config.Config) *Resource {
@@ -188,53 +190,171 @@ func (r *Resource) handleNginxPlusActionRequest(ctx context.Context, action *mpi
 
 	switch action.GetAction().(type) {
 	case *mpi.NGINXPlusAction_UpdateHttpUpstreamServers:
-		slog.DebugContext(ctx, "Updating http upstream servers",
-			"request", action.GetUpdateHttpUpstreamServers())
-		add, update, del, err := r.resourceService.UpdateHTTPUpstreams(ctx, instance,
-			action.GetUpdateHttpUpstreamServers().GetHttpUpstreamName(),
-			action.GetUpdateHttpUpstreamServers().GetServers())
-		if err != nil {
-			slog.ErrorContext(ctx, "Unable to update HTTP servers of upstream", "request",
-				action.GetUpdateHttpUpstreamServers(), "error", err)
-			resp := r.createDataPlaneResponse(correlationID, mpi.CommandResponse_COMMAND_STATUS_FAILURE,
-				"", instanceID, err.Error())
-			r.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
+		slog.DebugContext(ctx, "Updating http upstream servers", "request", action.GetUpdateHttpUpstreamServers())
+		r.handleUpdateHTTPUpstreamServers(ctx, action, instance)
+	case *mpi.NGINXPlusAction_GetHttpUpstreamServers:
+		slog.DebugContext(ctx, "Getting http upstream servers", "request", action.GetGetHttpUpstreamServers())
+		r.handleGetHTTPUpstreamServers(ctx, action, instance)
+	case *mpi.NGINXPlusAction_UpdateStreamServers:
+		slog.DebugContext(ctx, "Updating stream servers", "request", action.GetUpdateStreamServers())
+		r.handleUpdateStreamServers(ctx, action, instance)
+	case *mpi.NGINXPlusAction_GetStreamUpstreams:
+		slog.DebugContext(ctx, "Getting stream upstreams", "request", action.GetGetStreamUpstreams())
+		r.handleGetStreamUpstreams(ctx, instance)
+	case *mpi.NGINXPlusAction_GetUpstreams:
+		slog.DebugContext(ctx, "Getting upstreams", "request", action.GetGetUpstreams())
+		r.handleGetUpstreams(ctx, instance)
+	default:
+		slog.DebugContext(ctx, "NGINX Plus action not implemented yet")
+	}
+}
 
-			return
-		}
+// nolint: dupl
+func (r *Resource) handleUpdateStreamServers(ctx context.Context, action *mpi.NGINXPlusAction, instance *mpi.Instance) {
+	correlationID := logger.GetCorrelationID(ctx)
+	instanceID := instance.GetInstanceMeta().GetInstanceId()
 
-		slog.DebugContext(ctx, "Successfully updated http upstream servers", "http_upstream_name",
-			action.GetUpdateHttpUpstreamServers().GetHttpUpstreamName(), "add", len(add), "update", len(update),
-			"delete", len(del))
-		resp := r.createDataPlaneResponse(correlationID, mpi.CommandResponse_COMMAND_STATUS_OK,
-			"Successfully updated HTTP Upstreams", instanceID, "")
+	add, update, del, err := r.resourceService.UpdateStreamServers(ctx, instance,
+		action.GetUpdateStreamServers().GetUpstreamStreamName(), action.GetUpdateStreamServers().GetServers())
+	if err != nil {
+		slog.ErrorContext(ctx, "Unable to update stream servers of upstream", "request",
+			action.GetUpdateHttpUpstreamServers(), "error", err)
+		resp := r.createDataPlaneResponse(correlationID, mpi.CommandResponse_COMMAND_STATUS_FAILURE,
+			"", instanceID, err.Error())
+		r.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
 
+		return
+	}
+
+	slog.DebugContext(ctx, "Successfully updated stream upstream servers", "http_upstream_name",
+		action.GetUpdateHttpUpstreamServers().GetHttpUpstreamName(), "add", len(add), "update", len(update),
+		"delete", len(del))
+
+	resp := r.createDataPlaneResponse(correlationID, mpi.CommandResponse_COMMAND_STATUS_OK,
+		"Successfully updated stream upstream servers", instanceID, "")
+
+	r.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
+}
+
+// nolint: dupl
+func (r *Resource) handleGetStreamUpstreams(ctx context.Context, instance *mpi.Instance) {
+	correlationID := logger.GetCorrelationID(ctx)
+	instanceID := instance.GetInstanceMeta().GetInstanceId()
+	streamUpstreamsResponse := emptyResponse
+
+	streamUpstreams, err := r.resourceService.GetStreamUpstreams(ctx, instance)
+	if err != nil {
+		slog.ErrorContext(ctx, "Unable to get stream upstreams", "error", err)
+		resp := r.createDataPlaneResponse(correlationID, mpi.CommandResponse_COMMAND_STATUS_FAILURE,
+			"", instanceID, err.Error())
 		r.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
 
-	case *mpi.NGINXPlusAction_GetHttpUpstreamServers:
-		slog.DebugContext(ctx, "Getting http upstream servers", "request", action.GetGetHttpUpstreamServers())
-		upstreams, err := r.resourceService.GetUpstreams(ctx, instance,
-			action.GetGetHttpUpstreamServers().GetHttpUpstreamName())
-		if err != nil {
-			slog.ErrorContext(ctx, "Unable to get HTTP servers of upstream", "error", err)
-			resp := r.createDataPlaneResponse(correlationID, mpi.CommandResponse_COMMAND_STATUS_FAILURE,
-				"", instanceID, err.Error())
-			r.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
+		return
+	}
 
-			return
+	if streamUpstreams != nil {
+		streamUpstreamsJSON, jsonErr := json.Marshal(streamUpstreams)
+		if jsonErr != nil {
+			slog.ErrorContext(ctx, "Unable to marshal stream upstreams", "err", err)
 		}
+		streamUpstreamsResponse = string(streamUpstreamsJSON)
+	}
 
-		upstreamsJSON, err := json.Marshal(upstreams)
-		if err != nil {
-			slog.ErrorContext(ctx, "Unable to marshal http upstreams", "err", err)
+	resp := r.createDataPlaneResponse(correlationID, mpi.CommandResponse_COMMAND_STATUS_OK,
+		streamUpstreamsResponse, instanceID, "")
+
+	r.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
+}
+
+// nolint: dupl
+func (r *Resource) handleGetUpstreams(ctx context.Context, instance *mpi.Instance) {
+	correlationID := logger.GetCorrelationID(ctx)
+	instanceID := instance.GetInstanceMeta().GetInstanceId()
+	upstreamsResponse := emptyResponse
+
+	upstreams, err := r.resourceService.GetUpstreams(ctx, instance)
+	if err != nil {
+		slog.InfoContext(ctx, "Unable to get upstreams", "error", err)
+		resp := r.createDataPlaneResponse(correlationID, mpi.CommandResponse_COMMAND_STATUS_FAILURE,
+			"", instanceID, err.Error())
+		r.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
+
+		return
+	}
+
+	if upstreams != nil {
+		upstreamsJSON, jsonErr := json.Marshal(upstreams)
+		if jsonErr != nil {
+			slog.ErrorContext(ctx, "Unable to marshal upstreams", "err", err)
 		}
-		resp := r.createDataPlaneResponse(correlationID, mpi.CommandResponse_COMMAND_STATUS_OK,
-			string(upstreamsJSON), instanceID, "")
+		upstreamsResponse = string(upstreamsJSON)
+	}
+
+	resp := r.createDataPlaneResponse(correlationID, mpi.CommandResponse_COMMAND_STATUS_OK,
+		upstreamsResponse, instanceID, "")
+
+	r.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
+}
+
+// nolint: dupl
+func (r *Resource) handleUpdateHTTPUpstreamServers(ctx context.Context, action *mpi.NGINXPlusAction,
+	instance *mpi.Instance,
+) {
+	correlationID := logger.GetCorrelationID(ctx)
+	instanceID := instance.GetInstanceMeta().GetInstanceId()
 
+	add, update, del, err := r.resourceService.UpdateHTTPUpstreamServers(ctx, instance,
+		action.GetUpdateHttpUpstreamServers().GetHttpUpstreamName(),
+		action.GetUpdateHttpUpstreamServers().GetServers())
+	if err != nil {
+		slog.ErrorContext(ctx, "Unable to update HTTP servers of upstream", "request",
+			action.GetUpdateHttpUpstreamServers(), "error", err)
+		resp := r.createDataPlaneResponse(correlationID, mpi.CommandResponse_COMMAND_STATUS_FAILURE,
+			"", instanceID, err.Error())
 		r.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
-	default:
-		slog.DebugContext(ctx, "NGINX Plus action not implemented yet")
+
+		return
 	}
+
+	slog.DebugContext(ctx, "Successfully updated http upstream servers", "http_upstream_name",
+		action.GetUpdateHttpUpstreamServers().GetHttpUpstreamName(), "add", len(add), "update", len(update),
+		"delete", len(del))
+
+	resp := r.createDataPlaneResponse(correlationID, mpi.CommandResponse_COMMAND_STATUS_OK,
+		"Successfully updated HTTP Upstreams", instanceID, "")
+
+	r.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
+}
+
+func (r *Resource) handleGetHTTPUpstreamServers(ctx context.Context, action *mpi.NGINXPlusAction,
+	instance *mpi.Instance,
+) {
+	correlationID := logger.GetCorrelationID(ctx)
+	instanceID := instance.GetInstanceMeta().GetInstanceId()
+	upstreamsResponse := emptyResponse
+
+	upstreams, err := r.resourceService.GetHTTPUpstreamServers(ctx, instance,
+		action.GetGetHttpUpstreamServers().GetHttpUpstreamName())
+	if err != nil {
+		slog.ErrorContext(ctx, "Unable to get HTTP servers of upstream", "error", err)
+		resp := r.createDataPlaneResponse(correlationID, mpi.CommandResponse_COMMAND_STATUS_FAILURE,
+			"", instanceID, err.Error())
+		r.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
+
+		return
+	}
+
+	if upstreams != nil {
+		upstreamsJSON, jsonErr := json.Marshal(upstreams)
+		if jsonErr != nil {
+			slog.ErrorContext(ctx, "Unable to marshal http upstreams", "err", err)
+		}
+		upstreamsResponse = string(upstreamsJSON)
+	}
+	resp := r.createDataPlaneResponse(correlationID, mpi.CommandResponse_COMMAND_STATUS_OK,
+		upstreamsResponse, instanceID, "")
+
+	r.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
 }
 
 func (r *Resource) handleWriteConfigSuccessful(ctx context.Context, msg *bus.Message) {