Merge branch 'master' into fix-schema-registry-upgrade

.gitlab-ci.yml

@@ -0,0 +1,230 @@
+include:
+  - project: 'pluralsight/experience/gitlab-helpers'
+    file: '/salt-deploy/helper.yml'
+  - project: 'pluralsight/experience/gitlab-helpers'
+    file: '/helm-deploy/helper.yml'
+  - project: 'pluralsight/Technology/adapt/data-platform/gitlab-fragments/snyk'
+    ref: main
+    file: 'snyk-check.yml'
+
+image: openjdk:8
+
+variables:
+  ARTIFACT_NAME: hydra-publish
+  IMAGE_NAME: hydra
+  DOCKER_REGISTRY_URL: harbor.vnerd.com/library
+  DOCKER_REGISTRY_IMAGE: ${DOCKER_REGISTRY_URL}/${IMAGE_NAME}
+  ARTIFACTORY_REPOSITORY_URL: https://repository.vnerd.com/artifactory
+  ARTIFACTORY_REPOSITORY: bounded-context
+  ARTIFACTORY_ID: hydra/publish
+  ARTIFACTORY_ARTIFACT_VERSION: ${ARTIFACT_NAME}-${BUILD_VERSION}.tgz
+  ARTIFACTORY_PATH: ${ARTIFACTORY_ID}/${ARTIFACTORY_ARTIFACT_VERSION}
+  ARTIFACTORY_ARTIFACT_URL: ${ARTIFACTORY_REPOSITORY_URL}/${ARTIFACTORY_REPOSITORY}/${ARTIFACTORY_PATH}
+  BUILD_VERSION: 1.0.${CI_PIPELINE_IID}
+  BOUNDED_CONTEXT_DEV: adapt-dvs-dev
+  BOUNDED_CONTEXT_STAGING: adapt-dvs
+  BOUNDED_CONTEXT_PROD: adapt-dvs
+  ENV: ${CI_JOB_STAGE} 
+  SERVICE_NAME: "dev-hydra"
+  SLACK_ICON_EMOJI: ":gitlab:"
+  SLACK_CHANNEL: '#data-platform-alerts'
+  SLACK_MESSAGE: |
+    *[[SERVICE_NAME]]* deployed to *[[ENV]]*.
+    Version: *[[BUILD_VERSION]]*
+    [[PIPELINE_LINK]]
+  SLACK_USERNAME: 'GITLAB'
+  SNYK_SLACK_CHANNEL: '#dataops-snyk-vulnerability-alerts'
+  SNYK_SLACK_ICON_EMOJI: ":snyk:"
+  SNYK_ORG_ID: "d8094638-7a37-413f-b1b4-ad840fb9e239"
+  SNYK_PROJECT_ID: "d243a0e6-4ced-4efe-83fc-169d03b40cc7"
+  SNYK_PROJECT_URL: "https://app.snyk.io/org/ps-data-services"
+
+before_script:
+  - apt-get update -yqq
+  - apt-get install -yqq apt-transport-https apt-utils
+  - echo "deb https://repo.scala-sbt.org/scalasbt/debian /" | tee -a /etc/apt/sources.list.d/sbt.list
+  - mkdir -p /root/.gnupg
+  - gpg --recv-keys --no-default-keyring --keyring gnupg-ring:/etc/apt/trusted.gpg.d/scalasbt-release.gpg --keyserver hkp://keyserver.ubuntu.com:80 2EE0EA64E40A89B84B2DF73499E82A75642AC823
+  - chmod 644 /etc/apt/trusted.gpg.d/scalasbt-release.gpg
+  - apt-get update -yqq
+  - apt-get install -yqq sbt
+
+stages:
+  - build
+  - code-analysis
+  - package
+  - publish
+  - deploy_dev
+  - deploy_staging
+  - deploy_production
+  - notify
+
+build:
+  stage: build
+  cache: []
+  retry: 2
+  script:
+    - sbt clean compile test
+
+package:
+  stage: package
+  cache: []
+  script:
+    - sbt universal:packageBin
+    - ls ingest/target/universal/*.zip | xargs -I {} unzip {} -d ps-publish
+    - mv ps-publish/hydra-ingest*/* ps-publish
+    - rm -rf ps-publish/hydra-ingest*
+    - cd ps-publish/bin
+    - ls -la
+    - curl -O https://download.newrelic.com/newrelic/java-agent/newrelic-agent/4.4.0/newrelic-agent-4.4.0.jar
+    - curl -O https://download.newrelic.com/newrelic/java-agent/newrelic-agent/4.4.0/newrelic.yml
+    - cd ../../
+    - echo $BUILD_VERSION > VERSION
+    - tar czf ${ARTIFACT_NAME}-${BUILD_VERSION}.tgz --exclude=*.tmp --exclude=*.tgz --exclude=*.tgz.md5 VERSION ps-publish/
+    - echo "##teamcity[publishArtifacts '${ARTIFACT_NAME}-${BUILD_VERSION}.tgz']"
+    - md5sum ${ARTIFACT_NAME}-${BUILD_VERSION}.tgz > ${ARTIFACT_NAME}-${BUILD_VERSION}.tgz.md5
+    - echo "${ARTIFACTORY_ARTIFACT_URL}"
+    - |
+      curl \
+      -H "Authorization: Bearer ${NPM_TOKEN}" \
+      -X PUT ${ARTIFACTORY_ARTIFACT_URL} \
+      -T ${ARTIFACTORY_ARTIFACT_VERSION}
+  artifacts:
+    paths:
+      - ${ARTIFACT_NAME}-${BUILD_VERSION}.tgz
+      - ${ARTIFACT_NAME}-${BUILD_VERSION}.tgz.md5
+      - ps-publish/
+
+publish:
+  stage: publish
+  image: harbor.vnerd.com/proxy/library/docker:cli
+  tags:
+    - ps
+    - docker
+  needs:
+    - package
+  before_script:
+    - echo -n "$DOCKER_REGISTRY_PASSWORD" | docker login -u "$DOCKER_REGISTRY_USERNAME" --password-stdin "$DOCKER_REGISTRY_URL"
+  script:
+    - pwd
+    - ls -lh
+    - docker build -t ${IMAGE_NAME} -f Dockerfile.new .
+    - docker tag ${IMAGE_NAME} ${DOCKER_REGISTRY_IMAGE}:${BUILD_VERSION}-${CI_COMMIT_SHORT_SHA}
+    - docker push ${DOCKER_REGISTRY_IMAGE}:${BUILD_VERSION}-${CI_COMMIT_SHORT_SHA}
+    - docker tag ${IMAGE_NAME} ${DOCKER_REGISTRY_IMAGE}:latest
+    - docker push ${DOCKER_REGISTRY_IMAGE}:latest
+    - CURYEAR=$(date +%Y)
+    - CURMONTH=$(date +%-m)
+    - echo -n "${CURYEAR}.${CURMONTH}.${BUILD_VERSION}" > VERSION
+  variables:
+    DOCKER_HOST: tcp://localhost:2376
+    DOCKER_TLS_CERTDIR: "/certs"
+    DOCKER_CERT_PATH: "$DOCKER_TLS_CERTDIR/client"
+    DOCKER_TLS_VERIFY: 1
+
+.notify-all:
+  image: harbor.vnerd.com/library/ps-helm:latest
+  tags:
+    - ps
+    - docker 
+  script:
+    - ps-notify slack "$SLACK_MESSAGE" "$SLACK_CHANNEL"  --slack-username $SLACK_USERNAME --slack-icon-emoji $SLACK_ICON_EMOJI
+  variables:
+    NODE_ENV: $ENV
+
+deploy_dev:
+    stage: deploy_dev
+    needs: ["publish"]
+    extends: .helmDeploy
+    environment:
+      name: dev
+    before_script:
+      - sed -i -e "s/%IMAGE_TAG%/1.0.${CI_PIPELINE_IID}-${CI_COMMIT_SHORT_SHA}/g" $HELM_VALUES_FILE
+      - cat $HELM_VALUES_FILE
+      - echo 1.0.${CI_PIPELINE_IID}-${CI_COMMIT_SHORT_SHA}
+      - kubectl config set-cluster app-${CI_ENVIRONMENT_NAME%/*} --server=${HELM_K8S_CLUSTER_URL} --embed-certs --certificate-authority="$HELM_K8S_CLUSTER_CACERT"
+      - kubectl config set-credentials deploy --token=`echo ${HELM_DEPLOY_TOKEN} | base64 -d`
+      - kubectl config set-context deploy --cluster=app-${CI_ENVIRONMENT_NAME%/*} --namespace=${HELM_TARGET_NAMESPACE} --user=deploy
+      - kubectl config use-context deploy
+    variables:
+      HELM_CHART_NAME: ps-service
+      HELM_DEPLOY_TOKEN: ${HELM_DEV_DEPLOY_TOKEN}
+      K8S_CLUSTER_NAME: app-eks.eplur-staging.us-west-2
+      HELM_K8S_CLUSTER_URL: https://6C29C0073BB19BEF220B9437E6962AF2.gr7.us-west-2.eks.amazonaws.com
+      HELM_TARGET_NAMESPACE: ${BOUNDED_CONTEXT_DEV}
+      APPLICATION_ROLE: dev-hydra
+      HELM_VALUES_FILE: helm/eks-dev-values.yml
+
+slack:dev:
+  stage: notify
+  extends: .notify-all
+  when: on_success
+  needs: ['deploy_dev']
+  before_script:
+    - echo "Sending notification to slack"
+  variables:
+    ENV: "DEV Cluster"
+    SERVICE_NAME: "dev-hydra"
+
+
+deploy_staging: 
+  stage: deploy_staging
+  extends: .saltDeploy
+  when: manual
+  environment:
+      name: staging
+  before_script:
+    - cat $SALT_PILLAR
+    - sed -i "s/latest/$BUILD_VERSION/" $SALT_PILLAR
+    - cat $SALT_PILLAR  
+  variables:
+    CI_JOB_STAGE: stage
+    SALT_TARGET: roles:hydra-publish-msk
+    SALT_PASSWORD: ${LDAP_PASS}
+    SALT_ARGUMENTS: systemd-app
+    SALT_USERNAME: tcity-data-platform
+    SALT_KWARGS: 'failhard=true'
+    SALT_PILLAR: $CI_PROJECT_DIR/pillar_overrides.yaml
+    SALT_URL: https://saltmaster-stage.vnerd.com:8000
+
+slack:staging:
+  stage: notify
+  extends: .notify-all
+  when: on_success
+  needs: ['deploy_staging']
+  before_script:
+    - echo "Sending notification to slack"
+  variables:
+    ENV: "Staging Cluster"
+    SERVICE_NAME: "staging-hydra-publish"
+
+deploy_production: 
+  stage: deploy_production
+  extends: .saltDeploy
+  when: manual
+  environment:
+      name: production
+  before_script:
+    - cat $SALT_PILLAR
+    - sed -i "s/latest/$BUILD_VERSION/" $SALT_PILLAR
+    - cat $SALT_PILLAR  
+  variables:
+    CI_JOB_STAGE: stage
+    SALT_TARGET: roles:hydra-publish-msk
+    SALT_PASSWORD: ${LDAP_PASS}
+    SALT_ARGUMENTS: systemd-app
+    SALT_USERNAME: tcity-data-platform
+    SALT_KWARGS: 'failhard=true'
+    SALT_PILLAR: $CI_PROJECT_DIR/pillar_overrides.yaml
+    SALT_URL: https://saltmaster-production.vnerd.com:8000
+
+slack:production:
+  stage: notify
+  extends: .notify-all
+  when: on_success
+  needs: ['deploy_production']
+  before_script:
+    - echo "Sending notification to slack"
+  variables:
+    ENV: "Production Cluster"
+    SERVICE_NAME: "production-hydra-publish"

Dockerfile.new

@@ -2,7 +2,7 @@ FROM jelastic/jetty:9.4.49-openjdk-1.8.0_352
 
 USER root
 
-ENV JAVA_OPTS="-Xmx2G"
+ENV JAVA_OPTS="-Xmx2g"
 
 ENV CONTAINER_HTTP_PORT="8088"
 
@@ -12,4 +12,4 @@ EXPOSE 8088
 
 COPY ps-publish/ /ps-publish
 
-ENTRYPOINT ["/ps-publish/bin/hydra-ingest"]
+ENTRYPOINT ["/ps-publish/bin/hydra-ingest"]

avro/src/main/java/com/pluralsight/hydra/avro/JsonConverter.java

@@ -283,7 +283,17 @@ private Object typeConvert(Object value, String name, Schema schema) throws IOEx
             case STRING:
                 return value.toString();
             case ENUM:
-                boolean valid = schema.getEnumSymbols().contains(value.toString());
+                boolean valid;
+
+                if (value instanceof java.util.ArrayList) {
+                    valid = true;
+                    for (Object item : (ArrayList<Object>) value) {
+                        valid = valid && schema.getEnumSymbols().contains(item.toString());
+                    }
+                } else {
+                    valid = schema.getEnumSymbols().contains(value.toString());
+                }
+
                 if (!valid)
                     throw new IllegalArgumentException(value + " is not a valid symbol. Possible values are: " +
                             schema.getEnumSymbols() + ".");

common/src/main/scala/hydra/common/serdes/EnumEntryJsonFormat.scala

@@ -0,0 +1,20 @@
+package hydra.common.serdes
+
+import enumeratum.EnumEntry
+import spray.json.{DeserializationException, JsString, JsValue, RootJsonFormat}
+
+class EnumEntryJsonFormat[E <: EnumEntry](values: Seq[E]) extends RootJsonFormat[E] {
+
+  override def write(obj: E): JsValue = JsString(obj.entryName)
+
+  override def read(json: JsValue): E = json match {
+    case s: JsString => values.find(v => v.entryName == s.value).getOrElse(deserializationError(s))
+    case x => deserializationError(x)
+  }
+
+  private def deserializationError(value: JsValue) = {
+    val className = values.headOption.map(_.getClass.getEnclosingClass.getSimpleName).getOrElse("")
+    throw DeserializationException(
+      s"For '$className': Expected a value from enum $values instead of $value")
+  }
+}

common/src/main/scala/hydra/common/validation/AdditionalValidation.scala

@@ -0,0 +1,43 @@
+package hydra.common.validation
+
+import enumeratum.{Enum, EnumEntry}
+import vulcan.AvroNamespace
+
+import scala.collection.immutable
+
+@AvroNamespace("hydra.kafka.model")
+sealed trait AdditionalValidation extends EnumEntry
+sealed trait MetadataAdditionalValidation extends AdditionalValidation
+sealed trait SchemaAdditionalValidation extends AdditionalValidation
+
+// NOTE: Please note that any case object added here once must be retained throughout for schema to evolve.
+object MetadataAdditionalValidation extends Enum[MetadataAdditionalValidation] {
+
+  case object replacementTopics extends MetadataAdditionalValidation
+
+  override val values: immutable.IndexedSeq[MetadataAdditionalValidation] = findValues
+}
+
+// NOTE: Please note that any value added here once must be retained throughout for schema to evolve.
+object SchemaAdditionalValidation extends Enum[SchemaAdditionalValidation] {
+
+  case object defaultInRequiredField extends SchemaAdditionalValidation
+  case object timestampMillis extends SchemaAdditionalValidation
+
+  override val values: immutable.IndexedSeq[SchemaAdditionalValidation] = findValues
+}
+
+object AdditionalValidation {
+
+  lazy val allValidations: Option[List[AdditionalValidation]] =
+    Some(MetadataAdditionalValidation.values.toList ++ SchemaAdditionalValidation.values.toList)
+}
+
+class AdditionalValidationUtil(isExistingTopic: Boolean, currentAdditionalValidations: Option[List[AdditionalValidation]]) {
+
+  def pickValidations(): Option[List[AdditionalValidation]] =
+    if (isExistingTopic) currentAdditionalValidations else AdditionalValidation.allValidations
+
+  def isPresent(additionalValidation: AdditionalValidation): Boolean =
+    pickValidations().exists(_.contains(additionalValidation))
+}

core/src/main/scala/hydra/core/marshallers/HydraJsonSupport.scala

@@ -21,7 +21,9 @@ import java.util.UUID
 import akka.actor.ActorPath
 import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
 import akka.http.scaladsl.model.StatusCode
+import hydra.common.serdes.EnumEntryJsonFormat
 import hydra.common.util.Resource._
+import hydra.common.validation.AdditionalValidation
 import org.joda.time.DateTime
 import org.joda.time.format.ISODateTimeFormat
 import spray.json.{JsString, _}
@@ -157,7 +159,10 @@ trait HydraJsonSupport extends SprayJsonSupport with DefaultJsonProtocol {
 
   implicit val genericErrorFormat = jsonFormat2(GenericError)
 
-  implicit val topicCreationMetadataFormat = jsonFormat10(TopicMetadataRequest)
+  implicit val additionalValidationFormat: EnumEntryJsonFormat[AdditionalValidation] =
+    new EnumEntryJsonFormat[AdditionalValidation](Seq.empty)
+
+  implicit val topicCreationMetadataFormat = jsonFormat13(TopicMetadataRequest)
 
   implicit val genericSchemaFormat = jsonFormat2(GenericSchema)
 
@@ -170,12 +175,15 @@ case class TopicMetadataRequest(
     streamType: StreamType,
     derived: Boolean,
     deprecated: Option[Boolean],
+    replacementTopics: Option[List[String]],
+    previousTopics: Option[List[String]],
     dataClassification: String,
     subDataClassification: Option[String],
     contact: String,
     additionalDocumentation: Option[String],
     notes: Option[String],
-    notificationUrl: Option[String]
+    notificationUrl: Option[String],
+    additionalValidations: Option[List[AdditionalValidation]]
 ) {
   def updateDataClassification(dc: String): TopicMetadataRequest = this.copy(dataClassification = dc)