Skip to content

Commit 1ecfcad

Browse files
aphralGaphralG
committed
WIP plus api actions
1 parent 611e168 commit 1ecfcad

File tree

11 files changed

+1968
-368
lines changed

11 files changed

+1968
-368
lines changed

‎api/grpc/mpi/v1/command.pb.go‎

Lines changed: 477 additions & 266 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎api/grpc/mpi/v1/command.pb.validate.go‎

Lines changed: 462 additions & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎api/grpc/mpi/v1/command.proto‎

Lines changed: 21 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -204,8 +204,11 @@ message APIActionRequest {
204204
message NGINXPlusAction {
205205
// types of actions possible with NGINX Plus API
206206
oneof action {
207-
UpdateHTTPUpstreamServers update_http_upstream_servers = 2;
208-
GetHTTPUpstreamServers get_http_upstream_servers = 3;
207+
UpdateHTTPUpstreamServers update_http_upstream_servers = 1;
208+
GetHTTPUpstreamServers get_http_upstream_servers = 2;
209+
UpdateStreamServers update_stream_servers = 3;
210+
GetUpstreams get_upstreams = 4;
211+
GetStreamUpstreams get_stream_upstreams = 5;
209212
}
210213
}
211214

@@ -223,6 +226,22 @@ message GetHTTPUpstreamServers {
223226
string http_upstream_name = 1;
224227
}
225228

229+
// Update Upstream Stream Servers for an instance
230+
message UpdateStreamServers {
231+
// the name of the upstream stream
232+
string upstream_stream_name = 1;
233+
// a list of upstream stream servers
234+
repeated google.protobuf.Struct servers = 2;
235+
}
236+
237+
// Get Upstreams for an instance
238+
message GetUpstreams {
239+
}
240+
241+
// Get Stream Upstream Servers for an instance
242+
message GetStreamUpstreams {
243+
}
244+
226245
// Request an update on a particular command
227246
message CommandStatusRequest {}
228247

‎docs/proto/protos.md‎

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,8 @@
5252
- [DataPlaneResponse](#mpi-v1-DataPlaneResponse)
5353
- [FileServer](#mpi-v1-FileServer)
5454
- [GetHTTPUpstreamServers](#mpi-v1-GetHTTPUpstreamServers)
55+
- [GetStreamUpstreams](#mpi-v1-GetStreamUpstreams)
56+
- [GetUpstreams](#mpi-v1-GetUpstreams)
5557
- [HealthRequest](#mpi-v1-HealthRequest)
5658
- [HostInfo](#mpi-v1-HostInfo)
5759
- [Instance](#mpi-v1-Instance)
@@ -74,6 +76,7 @@
7476
- [UpdateDataPlaneStatusRequest](#mpi-v1-UpdateDataPlaneStatusRequest)
7577
- [UpdateDataPlaneStatusResponse](#mpi-v1-UpdateDataPlaneStatusResponse)
7678
- [UpdateHTTPUpstreamServers](#mpi-v1-UpdateHTTPUpstreamServers)
79+
- [UpdateStreamServers](#mpi-v1-UpdateStreamServers)
7780

7881
- [InstanceHealth.InstanceHealthStatus](#mpi-v1-InstanceHealth-InstanceHealthStatus)
7982
- [InstanceMeta.InstanceType](#mpi-v1-InstanceMeta-InstanceType)
@@ -797,6 +800,26 @@ Get HTTP Upstream Servers for an instance
797800

798801

799802

803+
<a name="mpi-v1-GetStreamUpstreams"></a>
804+
805+
### GetStreamUpstreams
806+
Get Stream Upstream Servers for an instance
807+
808+
809+
810+
811+
812+
813+
<a name="mpi-v1-GetUpstreams"></a>
814+
815+
### GetUpstreams
816+
Get Upstreams for an instance
817+
818+
819+
820+
821+
822+
800823
<a name="mpi-v1-HealthRequest"></a>
801824

802825
### HealthRequest
@@ -977,6 +1000,9 @@ Perform an action using the NGINX Plus API on an instance
9771000
| ----- | ---- | ----- | ----------- |
9781001
| update_http_upstream_servers | [UpdateHTTPUpstreamServers](#mpi-v1-UpdateHTTPUpstreamServers) | | |
9791002
| get_http_upstream_servers | [GetHTTPUpstreamServers](#mpi-v1-GetHTTPUpstreamServers) | | |
1003+
| update_stream_servers | [UpdateStreamServers](#mpi-v1-UpdateStreamServers) | | |
1004+
| get_upstreams | [GetUpstreams](#mpi-v1-GetUpstreams) | | |
1005+
| get_stream_upstreams | [GetStreamUpstreams](#mpi-v1-GetStreamUpstreams) | | |
9801006

9811007

9821008

@@ -1136,6 +1162,22 @@ Update HTTP Upstream Servers for an instance
11361162

11371163

11381164

1165+
1166+
<a name="mpi-v1-UpdateStreamServers"></a>
1167+
1168+
### UpdateStreamServers
1169+
Update Upstream Stream Servers for an instance
1170+
1171+
1172+
| Field | Type | Label | Description |
1173+
| ----- | ---- | ----- | ----------- |
1174+
| upstream_stream_name | [string](#string) | | the name of the upstream stream |
1175+
| servers | [google.protobuf.Struct](#google-protobuf-Struct) | repeated | a list of upstream stream servers |
1176+
1177+
1178+
1179+
1180+
11391181

11401182

11411183

‎internal/resource/resource_plugin.go‎

Lines changed: 149 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -188,53 +188,165 @@ func (r *Resource) handleNginxPlusActionRequest(ctx context.Context, action *mpi
188188

189189
switch action.GetAction().(type) {
190190
case *mpi.NGINXPlusAction_UpdateHttpUpstreamServers:
191-
slog.DebugContext(ctx, "Updating http upstream servers",
192-
"request", action.GetUpdateHttpUpstreamServers())
193-
add, update, del, err := r.resourceService.UpdateHTTPUpstreams(ctx, instance,
194-
action.GetUpdateHttpUpstreamServers().GetHttpUpstreamName(),
195-
action.GetUpdateHttpUpstreamServers().GetServers())
196-
if err != nil {
197-
slog.ErrorContext(ctx, "Unable to update HTTP servers of upstream", "request",
198-
action.GetUpdateHttpUpstreamServers(), "error", err)
199-
resp := r.createDataPlaneResponse(correlationID, mpi.CommandResponse_COMMAND_STATUS_FAILURE,
200-
"", instanceID, err.Error())
201-
r.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
191+
r.handleUpdateHTTPUpstreamServers(ctx, action, instance)
192+
case *mpi.NGINXPlusAction_GetHttpUpstreamServers:
193+
r.handleGetHTTPUpstreamServers(ctx, action, instance)
194+
case *mpi.NGINXPlusAction_UpdateStreamServers:
195+
r.handleUpdateStreamServers(ctx, action, instance)
196+
case *mpi.NGINXPlusAction_GetStreamUpstreams:
197+
r.handleGetStreamUpstreams(ctx, action, instance)
198+
case *mpi.NGINXPlusAction_GetUpstreams:
199+
r.handleGetUpstreams(ctx, action, instance)
200+
default:
201+
slog.DebugContext(ctx, "NGINX Plus action not implemented yet")
202+
}
203+
}
202204

203-
return
204-
}
205+
// nolint: dupl
206+
func (r *Resource) handleUpdateStreamServers(ctx context.Context, action *mpi.NGINXPlusAction, instance *mpi.Instance) {
207+
correlationID := logger.GetCorrelationID(ctx)
208+
instanceID := instance.GetInstanceMeta().GetInstanceId()
205209

206-
slog.DebugContext(ctx, "Successfully updated http upstream servers", "http_upstream_name",
207-
action.GetUpdateHttpUpstreamServers().GetHttpUpstreamName(), "add", len(add), "update", len(update),
208-
"delete", len(del))
209-
resp := r.createDataPlaneResponse(correlationID, mpi.CommandResponse_COMMAND_STATUS_OK,
210-
"Successfully updated HTTP Upstreams", instanceID, "")
210+
slog.DebugContext(ctx, "Updating stream servers", "request", action.GetUpdateStreamServers())
211211

212+
add, update, del, err := r.resourceService.UpdateStreamServers(ctx, instance,
213+
action.GetUpdateStreamServers().GetUpstreamStreamName(), action.GetUpdateStreamServers().GetServers())
214+
if err != nil {
215+
slog.ErrorContext(ctx, "Unable to update stream servers of upstream", "request",
216+
action.GetUpdateHttpUpstreamServers(), "error", err)
217+
resp := r.createDataPlaneResponse(correlationID, mpi.CommandResponse_COMMAND_STATUS_FAILURE,
218+
"", instanceID, err.Error())
212219
r.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
213220

214-
case *mpi.NGINXPlusAction_GetHttpUpstreamServers:
215-
slog.DebugContext(ctx, "Getting http upstream servers", "request", action.GetGetHttpUpstreamServers())
216-
upstreams, err := r.resourceService.GetUpstreams(ctx, instance,
217-
action.GetGetHttpUpstreamServers().GetHttpUpstreamName())
218-
if err != nil {
219-
slog.ErrorContext(ctx, "Unable to get HTTP servers of upstream", "error", err)
220-
resp := r.createDataPlaneResponse(correlationID, mpi.CommandResponse_COMMAND_STATUS_FAILURE,
221-
"", instanceID, err.Error())
222-
r.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
221+
return
222+
}
223223

224-
return
225-
}
224+
slog.DebugContext(ctx, "Successfully updated stream upstream servers", "http_upstream_name",
225+
action.GetUpdateHttpUpstreamServers().GetHttpUpstreamName(), "add", len(add), "update", len(update),
226+
"delete", len(del))
226227

227-
upstreamsJSON, err := json.Marshal(upstreams)
228-
if err != nil {
229-
slog.ErrorContext(ctx, "Unable to marshal http upstreams", "err", err)
230-
}
231-
resp := r.createDataPlaneResponse(correlationID, mpi.CommandResponse_COMMAND_STATUS_OK,
232-
string(upstreamsJSON), instanceID, "")
228+
resp := r.createDataPlaneResponse(correlationID, mpi.CommandResponse_COMMAND_STATUS_OK,
229+
"Successfully updated stream upstream servers", instanceID, "")
230+
231+
r.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
232+
}
233+
234+
// nolint: dupl
235+
func (r *Resource) handleGetStreamUpstreams(ctx context.Context, action *mpi.NGINXPlusAction, instance *mpi.Instance) {
236+
correlationID := logger.GetCorrelationID(ctx)
237+
instanceID := instance.GetInstanceMeta().GetInstanceId()
233238

239+
slog.DebugContext(ctx, "Getting Stream Upstreams", "request", action.GetUpdateStreamServers())
240+
241+
streamUpstreams, err := r.resourceService.GetStreamUpstreams(ctx, instance)
242+
if err != nil {
243+
slog.ErrorContext(ctx, "Unable to get stream upstreams", "error", err)
244+
resp := r.createDataPlaneResponse(correlationID, mpi.CommandResponse_COMMAND_STATUS_FAILURE,
245+
"", instanceID, err.Error())
234246
r.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
235-
default:
236-
slog.DebugContext(ctx, "NGINX Plus action not implemented yet")
247+
248+
return
249+
}
250+
251+
streamUpstreamsJSON, err := json.Marshal(streamUpstreams)
252+
if err != nil {
253+
slog.ErrorContext(ctx, "Unable to marshal stream upstreams", "err", err)
254+
}
255+
256+
resp := r.createDataPlaneResponse(correlationID, mpi.CommandResponse_COMMAND_STATUS_OK,
257+
string(streamUpstreamsJSON), instanceID, "")
258+
259+
r.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
260+
}
261+
262+
// nolint: dupl
263+
func (r *Resource) handleGetUpstreams(ctx context.Context, action *mpi.NGINXPlusAction, instance *mpi.Instance) {
264+
correlationID := logger.GetCorrelationID(ctx)
265+
instanceID := instance.GetInstanceMeta().GetInstanceId()
266+
267+
slog.DebugContext(ctx, "Getting upstreams", "request", action.GetUpdateStreamServers())
268+
269+
upstreams, err := r.resourceService.GetUpstreams(ctx, instance)
270+
if err != nil {
271+
slog.InfoContext(ctx, "Unable to get upstreams", "error", err)
272+
resp := r.createDataPlaneResponse(correlationID, mpi.CommandResponse_COMMAND_STATUS_FAILURE,
273+
"", instanceID, err.Error())
274+
r.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
275+
276+
return
277+
}
278+
279+
upstreamsJSON, err := json.Marshal(upstreams)
280+
if err != nil {
281+
slog.ErrorContext(ctx, "Unable to marshal upstreams", "err", err)
282+
}
283+
284+
resp := r.createDataPlaneResponse(correlationID, mpi.CommandResponse_COMMAND_STATUS_OK,
285+
string(upstreamsJSON), instanceID, "")
286+
287+
r.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
288+
}
289+
290+
// nolint: dupl
291+
func (r *Resource) handleUpdateHTTPUpstreamServers(ctx context.Context, action *mpi.NGINXPlusAction,
292+
instance *mpi.Instance,
293+
) {
294+
correlationID := logger.GetCorrelationID(ctx)
295+
instanceID := instance.GetInstanceMeta().GetInstanceId()
296+
297+
slog.DebugContext(ctx, "Updating http upstream servers", "request", action.GetUpdateHttpUpstreamServers())
298+
299+
add, update, del, err := r.resourceService.UpdateHTTPUpstreamServers(ctx, instance,
300+
action.GetUpdateHttpUpstreamServers().GetHttpUpstreamName(),
301+
action.GetUpdateHttpUpstreamServers().GetServers())
302+
if err != nil {
303+
slog.ErrorContext(ctx, "Unable to update HTTP servers of upstream", "request",
304+
action.GetUpdateHttpUpstreamServers(), "error", err)
305+
resp := r.createDataPlaneResponse(correlationID, mpi.CommandResponse_COMMAND_STATUS_FAILURE,
306+
"", instanceID, err.Error())
307+
r.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
308+
309+
return
237310
}
311+
312+
slog.DebugContext(ctx, "Successfully updated http upstream servers", "http_upstream_name",
313+
action.GetUpdateHttpUpstreamServers().GetHttpUpstreamName(), "add", len(add), "update", len(update),
314+
"delete", len(del))
315+
316+
resp := r.createDataPlaneResponse(correlationID, mpi.CommandResponse_COMMAND_STATUS_OK,
317+
"Successfully updated HTTP Upstreams", instanceID, "")
318+
319+
r.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
320+
}
321+
322+
func (r *Resource) handleGetHTTPUpstreamServers(ctx context.Context, action *mpi.NGINXPlusAction,
323+
instance *mpi.Instance,
324+
) {
325+
correlationID := logger.GetCorrelationID(ctx)
326+
instanceID := instance.GetInstanceMeta().GetInstanceId()
327+
328+
slog.DebugContext(ctx, "Getting http upstream servers", "request", action.GetGetHttpUpstreamServers())
329+
330+
upstreams, err := r.resourceService.GetHTTPUpstreamServers(ctx, instance,
331+
action.GetGetHttpUpstreamServers().GetHttpUpstreamName())
332+
if err != nil {
333+
slog.ErrorContext(ctx, "Unable to get HTTP servers of upstream", "error", err)
334+
resp := r.createDataPlaneResponse(correlationID, mpi.CommandResponse_COMMAND_STATUS_FAILURE,
335+
"", instanceID, err.Error())
336+
r.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
337+
338+
return
339+
}
340+
341+
upstreamsJSON, err := json.Marshal(upstreams)
342+
if err != nil {
343+
slog.ErrorContext(ctx, "Unable to marshal http upstreams", "err", err)
344+
}
345+
346+
resp := r.createDataPlaneResponse(correlationID, mpi.CommandResponse_COMMAND_STATUS_OK,
347+
string(upstreamsJSON), instanceID, "")
348+
349+
r.messagePipe.Process(ctx, &bus.Message{Topic: bus.DataPlaneResponseTopic, Data: resp})
238350
}
239351

240352
func (r *Resource) handleWriteConfigSuccessful(ctx context.Context, msg *bus.Message) {

0 commit comments

Comments
 (0)