Classify CH access denied (#4528)

Author: ilidemi

flow/alerting/classifier.go

@@ -250,6 +250,9 @@ var (
 	ErrorNotifyTooManyPartsError = ErrorClass{
 		Class: "NOTIFY_TOO_MANY_PARTS", action: NotifyUser,
 	}
+	ErrorNotifyClickHousePermissionsError = ErrorClass{
+		Class: "NOTIFY_CLICKHOUSE_PERMISSIONS_ERROR", action: NotifyUser,
+	}
 	// Catch-all for misc ClickHouse errors
 	ErrorNotifyClickHouseError = ErrorClass{
 		Class: "NOTIFY_CLICKHOUSE_ERROR", action: NotifyUser,
@@ -998,6 +1001,8 @@ func GetErrorClass(ctx context.Context, err error) (ErrorClass, ErrorInfo) {
 			if strings.HasSuffix(chException.Message, "is either DETACHED PERMANENTLY or was just created by another replica") {
 				return ErrorRetryRecoverable, chErrorInfo
 			}
+		case chproto.ErrAccessDenied:
+			return ErrorNotifyClickHousePermissionsError, chErrorInfo
 		case chproto.ErrUnsupportedMethod,
 			chproto.ErrIllegalColumn,
 			chproto.ErrDuplicateColumn,

flow/alerting/classifier_test.go

@@ -269,6 +269,21 @@ func TestClickHouseAccessEntityNotFoundErrorShouldBeRecoverable(t *testing.T) {
 	}
 }
 
+func TestClickHouseAccessDeniedErrorShouldBeNotifyPermissions(t *testing.T) {
+	err := &clickhouse.Exception{
+		Code:    int32(chproto.ErrAccessDenied),
+		Message: "user@example.com: Not enough privileges. To execute this query, it's necessary to have the grant READ ON S3",
+	}
+	errorClass, errInfo := GetErrorClass(t.Context(),
+		exceptions.NewNormalizationError(fmt.Errorf(
+			"failed to normalize records: failed to copy avro stages to destination: %w", err)))
+	assert.Equal(t, ErrorNotifyClickHousePermissionsError, errorClass, "Unexpected error class")
+	assert.Equal(t, ErrorInfo{
+		Source: ErrorSourceClickHouse,
+		Code:   strconv.Itoa(int(chproto.ErrAccessDenied)),
+	}, errInfo, "Unexpected error info")
+}
+
 func TestClickHousePushingToViewShouldBeMvError(t *testing.T) {
 	err := &clickhouse.Exception{
 		Code: int32(chproto.ErrCannotConvertType),