apikeyauthextension: Elasticsearch check error codes explicitly (#732)

* apikeyauthextension: check es auth error explicitly

Author: 1pkg

extension/apikeyauthextension/authenticator.go

@@ -26,6 +26,7 @@ import (
 	"errors"
 	"fmt"
 	"hash/fnv"
+	"net/http"
 	"strings"
 
 	"go.opentelemetry.io/collector/client"
@@ -263,6 +264,11 @@ func (a *authenticator) Authenticate(ctx context.Context, headers map[string][]s
 
 	hasPrivileges, username, err := a.hasPrivileges(ctx, authHeaderValue)
 	if err != nil {
+		if elasticsearchErr, ok := err.(*types.ElasticsearchError); ok {
+			if elasticsearchErr.Status == http.StatusUnauthorized || elasticsearchErr.Status == http.StatusForbidden {
+				return ctx, status.Error(codes.Unauthenticated, err.Error())
+			}
+		}
 		return ctx, fmt.Errorf(
 			"error checking privileges for API Key %q: %v", id, err,
 		)

extension/apikeyauthextension/authenticator_test.go

@@ -76,6 +76,19 @@ func TestAuthenticator(t *testing.T) {
 			}),
 			expectedErr: `error checking privileges for API Key "id": status: 400, failed: [a_type], reason: a_reason`,
 		},
+		"auth_error": {
+			handler: newCannedErrorHandler(types.ElasticsearchError{
+				ErrorCause: types.ErrorCause{
+					Type: "auth_reason",
+					Reason: func() *string {
+						reason := "auth_reason"
+						return &reason
+					}(),
+				},
+				Status: 401,
+			}),
+			expectedErr: `rpc error: code = Unauthenticated desc = status: 401, failed: [auth_reason], reason: auth_reason`,
+		},
 		"missing_privileges": {
 			handler:     newCannedHasPrivilegesHandler(hasprivileges.Response{HasAllRequested: false}),
 			expectedErr: `rpc error: code = PermissionDenied desc = API Key "id" unauthorized`,