Fix: meaningful errors during subscriptions (#235)

* fixed error propagation

---------

Signed-off-by: Artem Shcherbatiuk <[email protected]>

Author: vertex451

gateway/resolver/resolver.go

@@ -9,13 +9,12 @@ import (
 	"sort"
 	"strings"
 
-	"gopkg.in/yaml.v3"
-	v1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
-
 	"github.com/graphql-go/graphql"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/trace"
+	"gopkg.in/yaml.v3"
+	v1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime/schema"

gateway/resolver/subscription.go

@@ -7,6 +7,7 @@ import (
 	"strings"
 
 	"github.com/openmfp/golang-commons/sentry"
+	"github.com/pkg/errors"
 
 	v1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 
@@ -58,14 +59,14 @@ func (r *Service) runWatch(
 	labelSelector, err := getStringArg(p.Args, LabelSelectorArg, false)
 	if err != nil {
 		r.log.Error().Err(err).Msg("Failed to get label selector argument")
-		resultChannel <- errorResult("Failed to get label selector: " + err.Error())
+		resultChannel <- errors.Wrap(err, "failed to get label selector argument")
 		return
 	}
 
 	subscribeToAll, err := getBoolArg(p.Args, SubscribeToAllArg, false)
 	if err != nil {
 		r.log.Error().Err(err).Msg("Failed to get subscribeToAll argument")
-		resultChannel <- errorResult("Failed to get subscribeToAll: " + err.Error())
+		resultChannel <- errors.Wrap(err, "failed to get subscribeToAll argument")
 		return
 	}
 
@@ -84,7 +85,7 @@ func (r *Service) runWatch(
 		namespace, err = getStringArg(p.Args, NamespaceArg, isNamespaceRequired)
 		if err != nil {
 			r.log.Error().Err(err).Msg("Failed to get namespace argument")
-			resultChannel <- errorResult("Failed to get namespace: " + err.Error())
+			resultChannel <- errors.Wrap(err, "failed to get namespace argument")
 			return
 		}
 		if namespace != "" {
@@ -96,7 +97,7 @@ func (r *Service) runWatch(
 		selector, err := labels.Parse(labelSelector)
 		if err != nil {
 			r.log.Error().Err(err).Str("labelSelector", labelSelector).Msg("Invalid label selector")
-			resultChannel <- errorResult("Invalid label selector: " + err.Error())
+			resultChannel <- errors.Wrap(err, "invalid label selector")
 			return
 		}
 		opts = append(opts, client.MatchingLabelsSelector{Selector: selector})
@@ -107,7 +108,7 @@ func (r *Service) runWatch(
 		name, err = getStringArg(p.Args, NameArg, true)
 		if err != nil {
 			r.log.Error().Err(err).Msg("Failed to get name argument")
-			resultChannel <- errorResult("Failed to get name: " + err.Error())
+			resultChannel <- errors.Wrap(err, "failed to get name argument")
 			return
 		}
 		opts = append(opts, client.MatchingFields{"metadata.name": name})
@@ -116,7 +117,7 @@ func (r *Service) runWatch(
 	sortBy, err := getStringArg(p.Args, SortByArg, false)
 	if err != nil {
 		r.log.Error().Err(err).Msg("Failed to get sortBy argument")
-		resultChannel <- errorResult("Failed to get sortBy: " + err.Error())
+		resultChannel <- errors.Wrap(err, "failed to get sortBy argument")
 		return
 	}
 
@@ -134,8 +135,7 @@ func (r *Service) runWatch(
 
 		sentry.CaptureError(err, sentry.Tags{"namespace": namespace}, sentry.Extras{"gvk": gvk.String()})
 
-		resultChannel <- errorResult("Failed to start watch: " + err.Error())
-
+		resultChannel <- errors.Wrap(err, "failed to start watch")
 		return
 	}
 	defer watcher.Stop()
@@ -154,7 +154,8 @@ func (r *Service) runWatch(
 
 				sentry.CaptureError(err, sentry.Tags{"namespace": namespace})
 
-				continue
+				resultChannel <- errors.Wrap(err, "failed to cast event object to unstructured")
+				return
 			}
 			key := obj.GetNamespace() + "/" + obj.GetName()
 
@@ -175,7 +176,7 @@ func (r *Service) runWatch(
 
 						sentry.CaptureError(err, sentry.Tags{"namespace": namespace})
 
-						resultChannel <- errorResult("Failed to determine field changes: " + err.Error())
+						resultChannel <- errors.Wrap(err, "failed to determine field changed")
 						return
 					}
 					sendUpdate = changed
@@ -192,15 +193,16 @@ func (r *Service) runWatch(
 					if name != "" {
 						singleObj = previousObjects[namespace+"/"+name]
 					}
+
+					var data interface{}
+					if singleObj != nil { // object can be nil in case it is deleted
+						data = singleObj.Object
+					}
+
 					select {
 					case <-ctx.Done():
 						return
-					case resultChannel <- func() interface{} {
-						if singleObj == nil { // object will be nil in case it is deleted
-							return nil
-						}
-						return singleObj.Object
-					}():
+					case resultChannel <- data:
 					}
 				} else {
 					items := make([]unstructured.Unstructured, 0, len(previousObjects))
@@ -211,7 +213,7 @@ func (r *Service) runWatch(
 					err = validateSortBy(items, sortBy)
 					if err != nil {
 						r.log.Error().Err(err).Str(SortByArg, sortBy).Msg("Invalid sortBy field path")
-						resultChannel <- errorResult("Invalid sortBy field path: " + err.Error())
+						resultChannel <- errors.Wrap(err, "invalid sortBy field path")
 						return
 					}
 
@@ -336,8 +338,14 @@ func getFieldValue(obj *unstructured.Unstructured, fieldPath string) (interface{
 	return current, true, nil
 }
 
-func errorResult(msg string) map[string]interface{} {
-	return map[string]interface{}{
-		"error": msg,
+func CreateSubscriptionResolver(isSingle bool) graphql.FieldResolveFn {
+	return func(p graphql.ResolveParams) (interface{}, error) {
+		source := p.Source
+
+		if err, ok := source.(error); ok {
+			return nil, err
+		}
+
+		return source, nil
 	}
 }

gateway/schema/schema.go

@@ -252,22 +252,22 @@ func (g *Gateway) processSingleResource(
 
 	subscriptionSingular := strings.ToLower(fmt.Sprintf("%s_%s", gvk.Group, singular))
 	rootSubscriptionFields[subscriptionSingular] = &graphql.Field{
-		Type: addErrorFieldToGraphqlObject(resourceType),
+		Type: resourceType,
 		Args: itemArgsBuilder.
 			WithSubscribeToAll().
 			Complete(),
-		Resolve:     g.resolver.CommonResolver(),
+		Resolve:     resolver.CreateSubscriptionResolver(true),
 		Subscribe:   g.resolver.SubscribeItem(*gvk, resourceScope),
 		Description: fmt.Sprintf("Subscribe to changes of %s", singular),
 	}
 
 	subscriptionPlural := strings.ToLower(fmt.Sprintf("%s_%s", gvk.Group, plural))
 	rootSubscriptionFields[subscriptionPlural] = &graphql.Field{
-		Type: graphql.NewList(addErrorFieldToGraphqlObject(resourceType)),
+		Type: graphql.NewList(resourceType),
 		Args: listArgsBuilder.
 			WithSubscribeToAll().
 			Complete(),
-		Resolve:     g.resolver.CommonResolver(),
+		Resolve:     resolver.CreateSubscriptionResolver(false),
 		Subscribe:   g.resolver.SubscribeItems(*gvk, resourceScope),
 		Description: fmt.Sprintf("Subscribe to changes of %s", plural),
 	}
@@ -570,10 +570,3 @@ func sanitizeFieldName(name string) string {
 
 	return name
 }
-
-func addErrorFieldToGraphqlObject(obj *graphql.Object) *graphql.Object {
-	obj.AddFieldConfig("error", &graphql.Field{
-		Type: graphql.String,
-	})
-	return obj
-}