[views] Add /v1 views REST contract, controller, handler and validation

cluster/configs/src/main/java/com/linkedin/openhouse/cluster/configs/ClusterProperties.java

@@ -96,4 +96,12 @@ public class ClusterProperties {
   // string
   @Value("${cluster.tables.allowed-client-name-values:}")
   private List<String> allowedClientNameValues;
+
+  /**
+   * View SQL dialects this deployment accepts on the /v1 views API. A representation, and the
+   * source dialect naming one, is rejected unless its dialect is listed here. Defaults to Spark
+   * only; supporting another engine is a configuration change rather than a code change.
+   */
+  @Value("${cluster.tables.views.supported-dialects:spark}")
+  private List<String> viewsSupportedDialects;
 }

services/common/src/main/java/com/linkedin/openhouse/common/api/validator/ApiValidatorUtil.java

@@ -1,7 +1,12 @@
 package com.linkedin.openhouse.common.api.validator;
 
+import static com.linkedin.openhouse.common.api.validator.ValidatorConstants.ALPHA_NUM_UNDERSCORE_ERROR_MSG;
+import static com.linkedin.openhouse.common.api.validator.ValidatorConstants.ALPHA_NUM_UNDERSCORE_REGEX;
+
 import java.util.List;
 import javax.validation.ConstraintViolation;
+import javax.validation.Validator;
+import org.apache.commons.lang3.StringUtils;
 
 public final class ApiValidatorUtil {
   /**
@@ -45,4 +50,38 @@ public static void validatePageable(
               "sortBy : provided %s, does not support multiple sort fields or directions", sortBy));
     }
   }
+
+  /**
+   * Common method to run bean validation on a request object and append one formatted message per
+   * violation to the running list of validation failures.
+   *
+   * @param validator
+   * @param object
+   * @param validationFailures
+   */
+  public static <T> void collectViolations(
+      Validator validator, T object, List<String> validationFailures) {
+    for (ConstraintViolation<T> violation : validator.validate(object)) {
+      validationFailures.add(String.format("%s : %s", getField(violation), violation.getMessage()));
+    }
+  }
+
+  /**
+   * Common method to validate that an identifier is present and contains only the characters
+   * allowed by {@link ValidatorConstants#ALPHA_NUM_UNDERSCORE_REGEX}. At most one failure is
+   * reported: an absent identifier is not additionally reported as malformed.
+   *
+   * @param fieldName
+   * @param value
+   * @param validationFailures
+   */
+  public static void validateIdentifier(
+      String fieldName, String value, List<String> validationFailures) {
+    if (StringUtils.isEmpty(value)) {
+      validationFailures.add(String.format("%s : Cannot be empty", fieldName));
+    } else if (!value.matches(ALPHA_NUM_UNDERSCORE_REGEX)) {
+      validationFailures.add(
+          String.format("%s : provided %s, %s", fieldName, value, ALPHA_NUM_UNDERSCORE_ERROR_MSG));
+    }
+  }
 }

services/common/src/main/java/com/linkedin/openhouse/common/api/validator/ValidatorConstants.java

@@ -19,4 +19,24 @@ private ValidatorConstants() {}
       "Only alphanumerics, hyphen and underscore supported";
   public static final int MAX_ALLOWED_CLUSTERING_COLUMNS = 4;
   public static final String INITIAL_TABLE_VERSION = "INITIAL_VERSION";
+
+  /** The only view representation type accepted by the /v1 views API. */
+  public static final String SQL_VIEW_REPRESENTATION_TYPE = "sql";
+
+  /**
+   * Maximum length of a view or database identifier on the /v1 views API. Mirrors the
+   * {@code @Size(max = 128)} bean constraint on the request body identifiers so a path identifier
+   * cannot bypass the limit the body enforces.
+   */
+  public static final int MAX_VIEW_IDENTIFIER_LENGTH = 128;
+
+  /**
+   * Maximum size of a single view representation's SQL text, in UTF-8 bytes. Counted in bytes
+   * rather than characters because the limit protects storage and transport, and a {@code @Size}
+   * bean constraint would instead count UTF-16 characters and let a multibyte payload through.
+   */
+  public static final int MAX_VIEW_SQL_BYTES = 256 * 1024;
+
+  /** Maximum size of a view schema document, in UTF-8 bytes. See {@link #MAX_VIEW_SQL_BYTES}. */
+  public static final int MAX_VIEW_SCHEMA_BYTES = 512 * 1024;
 }

services/common/src/main/java/com/linkedin/openhouse/common/audit/ServiceAuditAspect.java

@@ -13,6 +13,8 @@
 import com.linkedin.openhouse.common.metrics.MetricsConstant;
 import java.nio.charset.StandardCharsets;
 import java.time.Instant;
+import java.util.Collections;
+import java.util.List;
 import javax.servlet.http.HttpServletRequest;
 import lombok.extern.slf4j.Slf4j;
 import org.aspectj.lang.ProceedingJoinPoint;
@@ -42,6 +44,13 @@ public class ServiceAuditAspect {
 
   @Autowired private AuditHandler<ServiceAuditEvent> serviceAuditHandler;
 
+  /**
+   * Redactors contributed by the running service, if any. Left empty when the service registers
+   * none, in which case the payload is audited exactly as it was sent.
+   */
+  @Autowired(required = false)
+  private List<ServiceAuditPayloadRedactor> payloadRedactors = Collections.emptyList();
+
   private static final MetricsReporter METRICS_REPORTER =
       MetricsReporter.of(MetricsConstant.SERVICE_AUDIT);
 
@@ -140,9 +149,12 @@ private ServiceAuditEvent buildServiceAuditEvent(
       if (wrapper != null) {
         String requestPayloadStr =
             new String(wrapper.getContentAsByteArray(), StandardCharsets.UTF_8);
-        requestPayload = JsonParser.parseString(requestPayloadStr);
+        requestPayload = redactSensitiveValues(request, JsonParser.parseString(requestPayloadStr));
       }
     } catch (Exception e) {
+      // Fail closed. A payload that could not be parsed or could not be redacted is dropped rather
+      // than audited raw, so a broken redactor cannot leak the values it was meant to remove.
+      requestPayload = null;
       log.error("Exception during parsing request payload:\n", e);
       METRICS_REPORTER.count(MetricsConstant.FAILED_PARSING_REQUEST_PAYLOAD);
     }
@@ -182,6 +194,25 @@ private ServiceAuditEvent buildServiceAuditEvent(
         .build();
   }
 
+  /**
+   * Apply every registered redactor that claims this request. With no redactor registered — the
+   * case for every service that has not contributed one — the parsed payload is returned unchanged,
+   * so existing audit payloads are byte-identical.
+   */
+  private JsonElement redactSensitiveValues(
+      HttpServletRequest request, JsonElement requestPayload) {
+    if (requestPayload == null || payloadRedactors == null) {
+      return requestPayload;
+    }
+    JsonElement redacted = requestPayload;
+    for (ServiceAuditPayloadRedactor redactor : payloadRedactors) {
+      if (redactor.appliesTo(request)) {
+        redacted = redactor.redact(redacted);
+      }
+    }
+    return redacted;
+  }
+
   private ServiceName getServiceNameFromRequestURI(String uri) {
     if (uri.startsWith("/jobs")) {
       return ServiceName.JOBS_SERVICE;

services/common/src/main/java/com/linkedin/openhouse/common/audit/ServiceAuditPayloadRedactor.java

@@ -0,0 +1,37 @@
+package com.linkedin.openhouse.common.audit;
+
+import com.google.gson.JsonElement;
+import javax.servlet.http.HttpServletRequest;
+
+/**
+ * Extension point that removes sensitive values from a request payload before {@link
+ * ServiceAuditAspect} writes it into a {@link
+ * com.linkedin.openhouse.common.audit.model.ServiceAuditEvent}.
+ *
+ * <p>The aspect audits the complete cached request body of every controller call, so a route whose
+ * body carries content that must not be retained has to opt out here. Only the mechanism lives in
+ * {@code services/common}: each service contributes its own beans and names its own fields, and a
+ * service that contributes none has its payload audited exactly as before.
+ *
+ * <p>Implementations are handed the parsed payload and must not mutate it. Returning a modified
+ * copy keeps a redactor from observing another redactor's half-rewritten tree, and keeps the
+ * decision to drop the payload entirely with the aspect.
+ */
+public interface ServiceAuditPayloadRedactor {
+
+  /**
+   * Marker written in place of a redacted value. Implementations replace the value and keep the
+   * key, so an auditor can still see that the field was present in the request.
+   */
+  String REDACTED_VALUE = "[REDACTED]";
+
+  /** @return whether this redactor owns {@code request}. */
+  boolean appliesTo(HttpServletRequest request);
+
+  /**
+   * @param requestPayload the parsed request body, which may be any {@link JsonElement} the caller
+   *     sent, including a non-object or {@link com.google.gson.JsonNull}.
+   * @return a redacted copy, or the argument unchanged when there is nothing to redact.
+   */
+  JsonElement redact(JsonElement requestPayload);
+}

services/common/src/main/java/com/linkedin/openhouse/common/exception/CodedApiException.java

@@ -0,0 +1,29 @@
+package com.linkedin.openhouse.common.exception;
+
+import org.springframework.http.HttpStatus;
+
+/**
+ * Base class for exceptions that already know the HTTP status they should map to.
+ *
+ * <p>This is deliberately a bare seam. It carries no error-code vocabulary of its own: a subclass
+ * in a downstream service owns whatever taxonomy it needs and reduces that taxonomy to an {@link
+ * HttpStatus} here. That keeps {@code services/common} free of any service-specific enum while
+ * still letting {@link com.linkedin.openhouse.common.exception.handler.OpenHouseExceptionHandler}
+ * map the exception to a response with a single handler.
+ *
+ * <p>The status is the only thing that reaches the wire. The error body shape is unchanged, so no
+ * subclass taxonomy is serialized.
+ */
+public abstract class CodedApiException extends RuntimeException {
+
+  protected CodedApiException(String message) {
+    super(message);
+  }
+
+  protected CodedApiException(String message, Throwable cause) {
+    super(message, cause);
+  }
+
+  /** @return the HTTP status this failure maps to. */
+  public abstract HttpStatus getHttpStatus();
+}

services/common/src/main/java/com/linkedin/openhouse/common/exception/handler/OpenHouseExceptionHandler.java

@@ -3,6 +3,7 @@
 import com.fasterxml.jackson.databind.exc.InvalidFormatException;
 import com.linkedin.openhouse.common.api.spec.ErrorResponseBody;
 import com.linkedin.openhouse.common.exception.AlreadyExistsException;
+import com.linkedin.openhouse.common.exception.CodedApiException;
 import com.linkedin.openhouse.common.exception.EntityConcurrentModificationException;
 import com.linkedin.openhouse.common.exception.InvalidSchemaEvolutionException;
 import com.linkedin.openhouse.common.exception.InvalidTableMetadataException;
@@ -100,6 +101,30 @@ protected ResponseEntity<ErrorResponseBody> handleToggleException(
     return buildResponseEntity(errorResponseBody);
   }
 
+  /**
+   * Generic mapping for any exception that already knows its own HTTP status. The status comes from
+   * {@link CodedApiException#getHttpStatus()}; the body is the existing unchanged {@link
+   * ErrorResponseBody}, so no service-specific error taxonomy reaches the wire.
+   *
+   * <p>Spring selects the most specific handler, so declaring this does not change the mapping of
+   * any exception already handled above.
+   */
+  @Hidden
+  @ExceptionHandler(CodedApiException.class)
+  protected ResponseEntity<ErrorResponseBody> handleCodedApiException(
+      CodedApiException codedApiException) {
+    HttpStatus httpStatus = codedApiException.getHttpStatus();
+    ErrorResponseBody errorResponseBody =
+        ErrorResponseBody.builder()
+            .status(httpStatus)
+            .error(httpStatus.getReasonPhrase())
+            .message(codedApiException.getMessage())
+            .stacktrace(getAbbreviatedStackTrace(codedApiException))
+            .cause(getExceptionCause(codedApiException))
+            .build();
+    return buildResponseEntity(errorResponseBody);
+  }
+
   /**
    * To customize behavior of handling {@link NoHandlerFoundException} one cannot rely on {@link
    * ExceptionHandler} as that causes ambiguity with {@link ResponseEntityExceptionHandler} and

services/common/src/test/java/com/linkedin/openhouse/common/api/validator/ApiValidatorUtilTest.java

@@ -0,0 +1,125 @@
+package com.linkedin.openhouse.common.api.validator;
+
+import static com.linkedin.openhouse.common.api.validator.ValidatorConstants.ALPHA_NUM_UNDERSCORE_ERROR_MSG;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Set;
+import javax.validation.ConstraintViolation;
+import javax.validation.Path;
+import javax.validation.Validator;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+public class ApiValidatorUtilTest {
+
+  /** Stand-in root bean, only used for the simple name that {@code getField} derives. */
+  private static final class SampleRequestBody {}
+
+  @Test
+  public void testValidateIdentifierAcceptsLegalIdentifier() {
+    List<String> validationFailures = new ArrayList<>();
+    ApiValidatorUtil.validateIdentifier("databaseId", "db_1", validationFailures);
+    Assertions.assertEquals(Collections.emptyList(), validationFailures);
+  }
+
+  @Test
+  public void testValidateIdentifierRejectsEmpty() {
+    List<String> validationFailures = new ArrayList<>();
+    ApiValidatorUtil.validateIdentifier("databaseId", "", validationFailures);
+    Assertions.assertEquals(
+        Collections.singletonList("databaseId : Cannot be empty"), validationFailures);
+  }
+
+  @Test
+  public void testValidateIdentifierRejectsNull() {
+    List<String> validationFailures = new ArrayList<>();
+    ApiValidatorUtil.validateIdentifier("tableId", null, validationFailures);
+    Assertions.assertEquals(
+        Collections.singletonList("tableId : Cannot be empty"), validationFailures);
+  }
+
+  @Test
+  public void testValidateIdentifierRejectsIllegalCharacters() {
+    List<String> validationFailures = new ArrayList<>();
+    ApiValidatorUtil.validateIdentifier("tableId", "t#1", validationFailures);
+    Assertions.assertEquals(
+        Collections.singletonList("tableId : provided t#1, " + ALPHA_NUM_UNDERSCORE_ERROR_MSG),
+        validationFailures);
+  }
+
+  @Test
+  public void testValidateIdentifierAppendsWithoutClearing() {
+    List<String> validationFailures = new ArrayList<>();
+    validationFailures.add("pre-existing failure");
+    ApiValidatorUtil.validateIdentifier("databaseId", "d$b", validationFailures);
+    Assertions.assertEquals(
+        Arrays.asList(
+            "pre-existing failure", "databaseId : provided d$b, " + ALPHA_NUM_UNDERSCORE_ERROR_MSG),
+        validationFailures);
+  }
+
+  @Test
+  public void testCollectViolationsFormatsEachViolation() {
+    SampleRequestBody requestBody = new SampleRequestBody();
+    Set<ConstraintViolation<SampleRequestBody>> violations = new LinkedHashSet<>();
+    violations.add(violation(requestBody, "tableId", "must not be empty"));
+    violations.add(violation(requestBody, "", "must be a consistent request"));
+
+    Validator validator = Mockito.mock(Validator.class);
+    Mockito.when(validator.validate(requestBody)).thenReturn(violations);
+
+    List<String> validationFailures = new ArrayList<>();
+    validationFailures.add("pre-existing failure");
+    ApiValidatorUtil.collectViolations(validator, requestBody, validationFailures);
+
+    Assertions.assertEquals(
+        Arrays.asList(
+            "pre-existing failure",
+            "SampleRequestBody.tableId : must not be empty",
+            "SampleRequestBody : must be a consistent request"),
+        validationFailures);
+  }
+
+  @Test
+  public void testCollectViolationsAddsNothingWhenBeanIsValid() {
+    SampleRequestBody requestBody = new SampleRequestBody();
+    Validator validator = Mockito.mock(Validator.class);
+    Mockito.when(validator.validate(requestBody)).thenReturn(Collections.emptySet());
+
+    List<String> validationFailures = new ArrayList<>();
+    ApiValidatorUtil.collectViolations(validator, requestBody, validationFailures);
+
+    Assertions.assertEquals(Collections.emptyList(), validationFailures);
+  }
+
+  @SuppressWarnings("unchecked")
+  private static ConstraintViolation<SampleRequestBody> violation(
+      SampleRequestBody rootBean, String propertyPath, String message) {
+    ConstraintViolation<SampleRequestBody> violation = Mockito.mock(ConstraintViolation.class);
+    Mockito.when(violation.getRootBean()).thenReturn(rootBean);
+    Mockito.when(violation.getPropertyPath()).thenReturn(path(propertyPath));
+    Mockito.when(violation.getMessage()).thenReturn(message);
+    return violation;
+  }
+
+  /** {@code getField} only reads the string form of a path, so an empty node list is enough. */
+  private static Path path(String value) {
+    return new Path() {
+      @Override
+      public Iterator<Node> iterator() {
+        return Collections.<Node>emptyList().iterator();
+      }
+
+      @Override
+      public String toString() {
+        return value;
+      }
+    };
+  }
+}