diff --git a/cmd/unbounded-net-controller/detail_responses.go b/cmd/unbounded-net-controller/detail_responses.go index 49c3f0e8d..9b4e3a8a7 100644 --- a/cmd/unbounded-net-controller/detail_responses.go +++ b/cmd/unbounded-net-controller/detail_responses.go @@ -28,7 +28,8 @@ func (m *nodeDetailRequests) Fail(nodeName, requestID, reason string) error { return errors.New("detail request node was deleted or replaced") } - if request.state == statusv1alpha1.NodeDetailComplete { + if request.state == statusv1alpha1.NodeDetailComplete || + (request.state == statusv1alpha1.NodeDetailUnavailable && request.message == reason) { return nil } @@ -56,6 +57,11 @@ func handleNodeDetailResponse(health *healthState, nodeName, requestID string, s return ack } + if failure != "" && status != nil { + ack.Reason = "detail response cannot contain both data and a collection error" + return ack + } + var err error if failure != "" { err = manager.Fail(nodeName, requestID, failure) diff --git a/cmd/unbounded-net-controller/detail_transport_test.go b/cmd/unbounded-net-controller/detail_transport_test.go index 573e9615b..205dddbf6 100644 --- a/cmd/unbounded-net-controller/detail_transport_test.go +++ b/cmd/unbounded-net-controller/detail_transport_test.go @@ -21,7 +21,7 @@ import ( statusv1alpha1 "github.com/Azure/unbounded/internal/net/status/v1alpha1" ) -func encodeDetailTransportMessage(t *testing.T, binary bool, requestID string) []byte { +func encodeDetailTransportMessage(t *testing.T, binary bool, requestID string, failure ...string) []byte { t.Helper() message := &statusproto.NodeStatusMessage{ @@ -47,6 +47,13 @@ func encodeDetailTransportMessage(t *testing.T, binary bool, requestID string) [ jsonMessage.DetailRequestID = requestID } + if len(failure) > 0 { + message.DetailError = failure[0] + message.Status = nil + jsonMessage.DetailError = failure[0] + jsonMessage.Status = nil + } + var ( data []byte err error @@ -212,6 +219,26 @@ func TestDetailWebSocketCommandAndResponse(t *testing.T) { if !manager.Result("node-a", request.RequestID).Details.ExpiresAt.Equal(expires) { t.Fatal("duplicate detail reply renewed TTL") } + + refresh := manager.Request("node-a", true) + if command := read(); command.DetailRequest == nil || command.DetailRequest.RequestID != refresh.RequestID { + t.Fatal("refresh command was not delivered") + } + + for range 2 { + if err := conn.Write(ctx, frameType, encodeDetailTransportMessage(t, binary, refresh.RequestID, "collection failed")); err != nil { + t.Fatal(err) + } + + if receipt := read(); receipt.Status != "ok" || receipt.DetailRequestID != refresh.RequestID { + t.Fatal("collection error receipt was not idempotently acknowledged") + } + } + + failed := manager.Result("node-a", refresh.RequestID) + if failed.State != statusv1alpha1.NodeDetailUnavailable || failed.Error != "collection failed" || failed.Details != nil { + t.Fatal("collection error did not become an explicit failed request") + } }) } } @@ -232,9 +259,9 @@ func TestDetailHTTPPollingCommandAndResponse(t *testing.T) { awaitDetailTransport(t, func() bool { _, ok := manager.Pending("node-a"); return ok }) - post := func(id string) *NodeStatusPushAck { + post := func(id string, failure ...string) *NodeStatusPushAck { t.Helper() - r := httptest.NewRequest(http.MethodPost, "/status/push", bytes.NewReader(encodeDetailTransportMessage(t, binary, id))) + r := httptest.NewRequest(http.MethodPost, "/status/push", bytes.NewReader(encodeDetailTransportMessage(t, binary, id, failure...))) r.Header.Set("Authorization", "Bearer "+token) if binary { @@ -266,6 +293,18 @@ func TestDetailHTTPPollingCommandAndResponse(t *testing.T) { if cached.Revision != publication.Revision || manager.Result("node-a", request.RequestID).Details == nil { t.Fatal("POST detail response changed routine state or failed to complete") } + + refresh := manager.Request("node-a", true) + for range 2 { + receipt := post(refresh.RequestID, "response exceeds the transport frame limit") + if receipt.Status != "ok" || receipt.DetailRequestID != refresh.RequestID { + t.Fatal("POST collection failure receipt was not acknowledged") + } + } + + if failed := manager.Result("node-a", refresh.RequestID); failed.State != statusv1alpha1.NodeDetailUnavailable || failed.Error == "" { + t.Fatal("POST collection failure did not terminate the request") + } }) } } diff --git a/cmd/unbounded-net-controller/server.go b/cmd/unbounded-net-controller/server.go index 79ecc9e27..965d6e39f 100644 --- a/cmd/unbounded-net-controller/server.go +++ b/cmd/unbounded-net-controller/server.go @@ -1247,6 +1247,10 @@ func handleStatusPushRequestWithSource(health *healthState, bodyBytes []byte, so envelope.Mode = "details" } + if envelope.DetailError != "" && envelope.Mode != "details" { + return NodeStatusPushAck{}, http.StatusBadRequest, fmt.Errorf("collection error requires a detail response") + } + if envelope.Summary != nil && envelope.Mode != "summary" { return NodeStatusPushAck{}, http.StatusBadRequest, fmt.Errorf("overview requires summary mode") } @@ -1290,7 +1294,7 @@ func handleStatusPushRequestWithSource(health *healthState, bodyBytes []byte, so return NodeStatusPushAck{Status: "error", DetailRequestID: envelope.DetailRequestID, Reason: "details cannot include a delta"}, http.StatusOK, nil } - return handleNodeDetailResponse(health, nodeName, envelope.DetailRequestID, envelope.Status, ""), http.StatusOK, nil + return handleNodeDetailResponse(health, nodeName, envelope.DetailRequestID, envelope.Status, envelope.DetailError), http.StatusOK, nil case "summary": if envelope.Summary == nil || envelope.Status != nil || envelope.Delta != nil || envelope.DetailRequestID != "" { return NodeStatusPushAck{}, http.StatusBadRequest, fmt.Errorf("summary must contain only overview data") @@ -1358,6 +1362,10 @@ func handleNodeStatusWSMessageWithSource(health *healthState, data []byte, sourc return "node_status_resync", NodeStatusPushAck{Status: "resync_required", Reason: "overview requires summary message type"} } + if message.DetailError != "" && message.Type != statusv1alpha1.NodeStatusDetailsType { + return "node_status_resync", NodeStatusPushAck{Status: "resync_required", Reason: "collection error requires a detail response"} + } + nodeName := message.NodeName if message.Status != nil && message.Status.NodeInfo.Name != "" { nodeName = message.Status.NodeInfo.Name @@ -1377,7 +1385,7 @@ func handleNodeStatusWSMessageWithSource(health *healthState, data []byte, sourc return "node_status_ack", NodeStatusPushAck{Status: "error", DetailRequestID: message.DetailRequestID, Reason: "details cannot include a delta"} } - return "node_status_ack", handleNodeDetailResponse(health, nodeName, message.DetailRequestID, message.Status, "") + return "node_status_ack", handleNodeDetailResponse(health, nodeName, message.DetailRequestID, message.Status, message.DetailError) case statusv1alpha1.NodeStatusSummaryType: if message.Summary == nil || message.Status != nil || message.Delta != nil || message.DetailRequestID != "" { return "node_status_resync", NodeStatusPushAck{Status: "resync_required", Reason: "summary must contain only overview data"} diff --git a/cmd/unbounded-net-controller/status_overview_ingestion_test.go b/cmd/unbounded-net-controller/status_overview_ingestion_test.go index 87cddf225..05e49ee46 100644 --- a/cmd/unbounded-net-controller/status_overview_ingestion_test.go +++ b/cmd/unbounded-net-controller/status_overview_ingestion_test.go @@ -27,6 +27,7 @@ func submitOverview(t *testing.T, channel string, health *healthState, message * } else { envelope := NodeStatusWSMessage{ Type: message.Type, NodeName: message.NodeName, DetailRequestID: message.DetailRequestId, + DetailError: message.DetailError, } if message.Summary != nil { overview := protoToNodeOverview(message.Summary) @@ -143,6 +144,7 @@ func TestOverviewIngestionRejectsInvalidEnvelopes(t *testing.T) { {"full mixed with summary", func(m *statusproto.NodeStatusMessage) { m.Status = &statusproto.NodeStatusFull{} }}, {"delta mixed with summary", func(m *statusproto.NodeStatusMessage) { m.Delta = &statusproto.NodeStatusDelta{} }}, {"detail correlation on summary", func(m *statusproto.NodeStatusMessage) { m.DetailRequestId = "request" }}, + {"collection error on summary", func(m *statusproto.NodeStatusMessage) { m.DetailError = "failed" }}, {"summary in full", func(m *statusproto.NodeStatusMessage) { m.Type = "node_status_full" }}, } { t.Run(channel+"/"+tc.name, func(t *testing.T) { diff --git a/cmd/unbounded-net-controller/status_proto.go b/cmd/unbounded-net-controller/status_proto.go index 5969e3789..635ebc611 100644 --- a/cmd/unbounded-net-controller/status_proto.go +++ b/cmd/unbounded-net-controller/status_proto.go @@ -483,6 +483,10 @@ func handleProtoWSMessage(health *healthState, decoded *decodedProtoWSMessage, s msg := &decoded.message nodeName := decoded.nodeName + if msg.DetailError != "" && msg.Type != statusv1alpha1.NodeStatusDetailsType { + return "node_status_resync", NodeStatusPushAck{Status: "resync_required", Reason: "collection error requires a detail response"} + } + if msg.Summary != nil && msg.Type != statusv1alpha1.NodeStatusSummaryType { return "node_status_resync", NodeStatusPushAck{Status: "resync_required", Reason: "overview requires summary message type"} } @@ -505,7 +509,7 @@ func handleProtoWSMessage(health *healthState, decoded *decodedProtoWSMessage, s status = &full } - return "node_status_ack", handleNodeDetailResponse(health, nodeName, msg.DetailRequestId, status, "") + return "node_status_ack", handleNodeDetailResponse(health, nodeName, msg.DetailRequestId, status, msg.DetailError) case statusv1alpha1.NodeStatusSummaryType: if msg.Summary == nil || msg.Status != nil || msg.Delta != nil || msg.DetailRequestId != "" { return "node_status_resync", NodeStatusPushAck{Status: "resync_required", Reason: "summary must contain only overview data"} @@ -572,6 +576,10 @@ func handleProtoPushRequest(health *healthState, bodyBytes []byte, source string ack = NodeStatusPushAck{Status: "ok"} + if msg.DetailError != "" && msg.Type != statusv1alpha1.NodeStatusDetailsType { + return NodeStatusPushAck{}, 400, fmt.Errorf("collection error requires a detail response") + } + if msg.Summary != nil && msg.Type != statusv1alpha1.NodeStatusSummaryType { return NodeStatusPushAck{}, 400, fmt.Errorf("overview requires summary message type") } @@ -594,7 +602,7 @@ func handleProtoPushRequest(health *healthState, bodyBytes []byte, source string status = &full } - return handleNodeDetailResponse(health, nodeName, msg.DetailRequestId, status, ""), 200, nil + return handleNodeDetailResponse(health, nodeName, msg.DetailRequestId, status, msg.DetailError), 200, nil case statusv1alpha1.NodeStatusSummaryType: if msg.Summary == nil || msg.Status != nil || msg.Delta != nil || msg.DetailRequestId != "" { return NodeStatusPushAck{}, 400, fmt.Errorf("summary must contain only overview data") diff --git a/cmd/unbounded-net-controller/status_types.go b/cmd/unbounded-net-controller/status_types.go index 2c3cf41d9..a5d62a13d 100644 --- a/cmd/unbounded-net-controller/status_types.go +++ b/cmd/unbounded-net-controller/status_types.go @@ -85,6 +85,7 @@ type NodeStatusPushEnvelope struct { Delta map[string]json.RawMessage `json:"delta,omitempty"` Summary *statusv1alpha1.NodeStatusOverview `json:"summary,omitempty"` DetailRequestID string `json:"detailRequestId,omitempty"` + DetailError string `json:"detailError,omitempty"` SupportsDetails bool `json:"supportsDetails,omitempty"` } diff --git a/internal/net/status/detail_error_test.go b/internal/net/status/detail_error_test.go new file mode 100644 index 000000000..d041fd7b3 --- /dev/null +++ b/internal/net/status/detail_error_test.go @@ -0,0 +1,53 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package status + +import ( + "encoding/json" + "testing" + + "google.golang.org/protobuf/proto" + + statusproto "github.com/Azure/unbounded/internal/net/status/proto" + "github.com/Azure/unbounded/internal/net/status/v1alpha1" +) + +func TestCorrelatedDetailError(t *testing.T) { + message := &statusproto.NodeStatusMessage{ + Type: v1alpha1.NodeStatusDetailsType, NodeName: "node", + DetailRequestId: "request", DetailError: "detail response exceeds transport limit", + } + + data, err := proto.Marshal(message) + if err != nil { + t.Fatal(err) + } + + var decoded statusproto.NodeStatusMessage + if err := proto.Unmarshal(data, &decoded); err != nil { + t.Fatal(err) + } + + if !proto.Equal(message, &decoded) || decoded.Status != nil || decoded.BaseRevision != 0 { + t.Fatalf("failure lost correlation or became a publication: %v", &decoded) + } + + jsonMessage := v1alpha1.NodeStatusMessage{ + Type: message.Type, NodeName: message.NodeName, DetailRequestID: message.DetailRequestId, DetailError: message.DetailError, + } + + data, err = json.Marshal(jsonMessage) + if err != nil { + t.Fatal(err) + } + + var jsonDecoded v1alpha1.NodeStatusMessage + if err := json.Unmarshal(data, &jsonDecoded); err != nil { + t.Fatal(err) + } + + if jsonDecoded.DetailError != message.DetailError || jsonDecoded.DetailRequestID != message.DetailRequestId || jsonDecoded.Status != nil { + t.Fatalf("JSON failure lost facts: %+v", jsonDecoded) + } +} diff --git a/internal/net/status/proto/status.pb.go b/internal/net/status/proto/status.pb.go index 79eff0d73..ef23200f3 100644 --- a/internal/net/status/proto/status.pb.go +++ b/internal/net/status/proto/status.pb.go @@ -35,6 +35,7 @@ type NodeStatusMessage struct { Summary *NodeStatusOverview `protobuf:"bytes,6,opt,name=summary,proto3" json:"summary,omitempty"` // complete overview, including on resync DetailRequestId string `protobuf:"bytes,7,opt,name=detail_request_id,json=detailRequestId,proto3" json:"detail_request_id,omitempty"` // correlates one-shot details in status, never a revision SupportsDetails bool `protobuf:"varint,8,opt,name=supports_details,json=supportsDetails,proto3" json:"supports_details,omitempty"` + DetailError string `protobuf:"bytes,9,opt,name=detail_error,json=detailError,proto3" json:"detail_error,omitempty"` // correlated one-shot failure; status is unset unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -125,6 +126,13 @@ func (x *NodeStatusMessage) GetSupportsDetails() bool { return false } +func (x *NodeStatusMessage) GetDetailError() string { + if x != nil { + return x.DetailError + } + return "" +} + // NodeStatusAck is the acknowledgment returned by the controller for push updates. type NodeStatusAck struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -2061,7 +2069,7 @@ var File_status_proto protoreflect.FileDescriptor const file_status_proto_rawDesc = "" + "\n" + - "\fstatus.proto\x12\x16unboundednet.status.v1\"\x85\x03\n" + + "\fstatus.proto\x12\x16unboundednet.status.v1\"\xa8\x03\n" + "\x11NodeStatusMessage\x12\x12\n" + "\x04type\x18\x01 \x01(\tR\x04type\x12\x1b\n" + "\tnode_name\x18\x02 \x01(\tR\bnodeName\x12#\n" + @@ -2070,7 +2078,8 @@ const file_status_proto_rawDesc = "" + "\x05delta\x18\x05 \x01(\v2'.unboundednet.status.v1.NodeStatusDeltaR\x05delta\x12D\n" + "\asummary\x18\x06 \x01(\v2*.unboundednet.status.v1.NodeStatusOverviewR\asummary\x12*\n" + "\x11detail_request_id\x18\a \x01(\tR\x0fdetailRequestId\x12)\n" + - "\x10supports_details\x18\b \x01(\bR\x0fsupportsDetails\"\xaf\x02\n" + + "\x10supports_details\x18\b \x01(\bR\x0fsupportsDetails\x12!\n" + + "\fdetail_error\x18\t \x01(\tR\vdetailError\"\xaf\x02\n" + "\rNodeStatusAck\x12\x16\n" + "\x06status\x18\x01 \x01(\tR\x06status\x12\x1a\n" + "\brevision\x18\x02 \x01(\x04R\brevision\x12\x16\n" + diff --git a/internal/net/status/proto/status.proto b/internal/net/status/proto/status.proto index 867fafa46..f276bccb0 100644 --- a/internal/net/status/proto/status.proto +++ b/internal/net/status/proto/status.proto @@ -15,6 +15,7 @@ message NodeStatusMessage { NodeStatusOverview summary = 6; // complete overview, including on resync string detail_request_id = 7; // correlates one-shot details in status, never a revision bool supports_details = 8; + string detail_error = 9; // correlated one-shot failure; status is unset } // NodeStatusAck is the acknowledgment returned by the controller for push updates. diff --git a/internal/net/status/v1alpha1/messages.go b/internal/net/status/v1alpha1/messages.go index 3ec66ced8..e14261a9c 100644 --- a/internal/net/status/v1alpha1/messages.go +++ b/internal/net/status/v1alpha1/messages.go @@ -31,6 +31,7 @@ type NodeStatusMessage struct { Summary *NodeStatusOverview `json:"summary,omitempty"` DetailRequestID string `json:"detailRequestId,omitempty"` SupportsDetails bool `json:"supportsDetails,omitempty"` + DetailError string `json:"detailError,omitempty"` } // NodeStatusAck is shared by HTTP responses and WebSocket ACK/command data.