Merge pull request #1529 from spaceuptech/v0.21.0

v0.21.0

Author: YourTechBud

dbevents/Dockerfile

@@ -23,6 +23,7 @@ COPY --from=stage0 --chown=demiourgos728:root /1/opt/docker /opt/docker
 COPY --from=stage0 --chown=demiourgos728:root /2/opt/docker /opt/docker
 COPY --from=stage1 --chown=demiourgos728:root /build/app /usr/local/bin/conn-string-parser
 COPY --chown=demiourgos728:root src/main/resources/application.conf /config/application.conf
+RUN chmod -R 0777 /opt/docker
 USER 1001:0
 ENTRYPOINT ["/opt/docker/bin/db-events-soruce"]
 CMD []

dbevents/build.sbt

@@ -1,6 +1,6 @@
 name := "db-events-soruce"
 
-version := "0.1.0"
+version := "0.2.0"
 
 scalaVersion := "2.13.1"
 

dbevents/src/main/scala/com/spaceuptech/dbevents/database/Utils.scala

@@ -1,21 +1,22 @@
 package com.spaceuptech.dbevents.database
 
-import java.nio.ByteBuffer
-import java.util.{Calendar, Properties}
-import java.util.concurrent.{Callable, ExecutorService}
-import java.util.function.Consumer
-
 import akka.actor.typed.ActorRef
 import com.mongodb.client.model.changestream.{ChangeStreamDocument, FullDocument, OperationType}
 import com.mongodb.{MongoClient, MongoClientURI}
 import com.spaceuptech.dbevents.database.Database.ChangeRecord
 import com.spaceuptech.dbevents.{DatabaseSource, Global}
 import io.debezium.engine.format.Json
 import io.debezium.engine.{ChangeEvent, DebeziumEngine}
-import org.bson.{BsonDocument, Document}
+import org.bson.{BsonDocument, BsonType, BsonValue, Document}
 import org.json4s._
 import org.json4s.jackson.JsonMethods._
 
+import java.nio.ByteBuffer
+import java.time.{Instant, OffsetDateTime, ZoneId}
+import java.util.concurrent.{Callable, ExecutorService}
+import java.util.function.Consumer
+import java.util.{Calendar, Properties}
+
 object Utils {
   def startMongoWatcher(projectId: String, dbAlias: String, conn: String, dbName: String, executorService: ExecutorService, actor: ActorRef[Database.Command]): MongoStatus = {
     // Make a mongo client
@@ -58,7 +59,7 @@ object Utils {
               payload = ChangeRecordPayload(
                 op = "c",
                 before = None,
-                after = Some(mongoDocumentToMap(doc.getFullDocument)),
+                after = Some(mongoBsonValueToValue(doc.getFullDocument.toBsonDocument(classOf[BsonValue], MongoClient.getDefaultCodecRegistry)).asInstanceOf[Map[String, Any]]),
                 source = getMongoSource(projectId, dbAlias, doc)
               ),
               project = projectId,
@@ -72,7 +73,7 @@ object Utils {
               payload = ChangeRecordPayload(
                 op = "u",
                 before = Option(mongoDocumentKeyToMap(doc.getDocumentKey)),
-                after = Some(mongoDocumentToMap(doc.getFullDocument)),
+                after = Some(mongoBsonValueToValue(doc.getFullDocument.toBsonDocument(classOf[BsonValue], MongoClient.getDefaultCodecRegistry)).asInstanceOf[Map[String, Any]]),
                 source = getMongoSource(projectId, dbAlias, doc)
               ),
               project = projectId,
@@ -130,7 +131,7 @@ object Utils {
     BsonDocument.parse(new String(data.array(), "UTF-8"))
   }
 
-  def mongoDocumentKeyToMap(find: BsonDocument): Map[String, Any] =  {
+  def mongoDocumentKeyToMap(find: BsonDocument): Map[String, Any] = {
     var id: String = ""
     val field = find.get("_id")
 
@@ -145,21 +146,50 @@ object Utils {
     Map("_id" -> id)
   }
 
-  def mongoDocumentToMap(doc: Document): Map[String, Any] =  {
-    implicit val formats: DefaultFormats.type = org.json4s.DefaultFormats
-
-    // Convert to json object
-    val jsonString = doc.toJson
-    var m = parse(jsonString).extract[Map[String, Any]]
-
-    // See _id is an object id
-    try {
-      m += "_id" -> doc.getObjectId("_id").toHexString
-    } catch {
-      case _: Throwable =>
+  def mongoBsonValueToValue(value: BsonValue): Any = {
+    // Skip if value is null
+    if (value == null) return null
+
+    value.getBsonType match {
+      case BsonType.INT32 =>
+        value.asInt32().getValue
+      case BsonType.INT64 =>
+        value.asInt64().getValue
+      case BsonType.OBJECT_ID =>
+        value.asObjectId().getValue.toHexString
+      case BsonType.DATE_TIME =>
+        OffsetDateTime.ofInstant(Instant.ofEpochMilli(value.asDateTime().getValue), ZoneId.systemDefault()).toString
+      case BsonType.TIMESTAMP =>
+        OffsetDateTime.ofInstant(Instant.ofEpochMilli(value.asTimestamp().getValue), ZoneId.systemDefault()).toString
+      case BsonType.DOCUMENT =>
+        val doc = value.asDocument().toBsonDocument(classOf[BsonDocument], MongoClient.getDefaultCodecRegistry)
+        var obj: Map[String, Any] = Map.empty
+        doc.keySet().forEach(key => {
+          obj += (key -> mongoBsonValueToValue(doc.get(key)))
+        })
+        obj
+      case BsonType.ARRAY =>
+        val arr = value.asArray().getValues
+        var op: Array[Any] = Array.empty
+
+        arr.forEach(value => {
+          op = op :+ mongoBsonValueToValue(value)
+        })
+
+        op
+      case BsonType.BOOLEAN =>
+        value.asBoolean().getValue
+      case BsonType.STRING =>
+        value.asString().getValue
+      case BsonType.DOUBLE =>
+        value.asDouble().getValue
+      case BsonType.DECIMAL128 =>
+        value.asDecimal128().doubleValue()
+      case BsonType.BINARY =>
+        value.asBinary().getData
+      case _ =>
+        value.toString
     }
-
-    m
   }
 
   def getMongoSource(projectId: String, dbAlias: String, doc: ChangeStreamDocument[Document]): ChangeRecordPayloadSource = {
@@ -232,6 +262,24 @@ object Utils {
     props
   }
 
+  def getOffsetStorageClass: String = {
+    Global.storageType match {
+      case "k8s" => "com.spaceuptech.dbevents.database.KubeOffsetBackingStore"
+      case _ => "org.apache.kafka.connect.storage.FileOffsetBackingStore"
+    }
+  }
+
+  def getDatabaseHistoryStorageClass: String = {
+    Global.storageType match {
+      case "k8s" => "com.spaceuptech.dbevents.database.KubeDatabaseHistory"
+      case _ => "io.debezium.relational.history.FileDatabaseHistory"
+    }
+  }
+
+  def generateConnectorName(source: DatabaseSource): String = {
+    s"${source.project}_${source.dbAlias}"
+  }
+
   def prepareSQLServerConfig(source: DatabaseSource): Properties = {
     val name = generateConnectorName(source)
 
@@ -247,16 +295,14 @@ object Utils {
     props.setProperty("database.hostname", source.config.getOrElse("host", "localhost"))
     props.setProperty("database.port", source.config.getOrElse("port", "1433"))
     props.setProperty("database.user", source.config.getOrElse("user", "sa"))
-    props.setProperty("database.password", source.config.getOrElse("password", "mypassword"))
+    props.setProperty("database.password", source.config.getOrElse("password", "password"))
     props.setProperty("database.dbname", source.config.getOrElse("db", "test"))
     props.setProperty("database.server.name", s"${generateConnectorName(source)}_connector")
     props.setProperty("database.history", getDatabaseHistoryStorageClass)
     props.setProperty("database.history.file.filename", s"./dbevents-dbhistory-$name.dat")
     props
   }
 
-
-
   def preparePostgresConfig(source: DatabaseSource): Properties = {
     val name = generateConnectorName(source)
 
@@ -284,22 +330,4 @@ object Utils {
 
     props
   }
-
-  def getOffsetStorageClass: String = {
-    Global.storageType match {
-      case "k8s" => "com.spaceuptech.dbevents.database.KubeOffsetBackingStore"
-      case _ => "org.apache.kafka.connect.storage.FileOffsetBackingStore"
-    }
-  }
-
-  def getDatabaseHistoryStorageClass: String = {
-    Global.storageType match {
-      case "k8s" => "com.spaceuptech.dbevents.database.KubeDatabaseHistory"
-      case _ => "io.debezium.relational.history.FileDatabaseHistory"
-    }
-  }
-
-  def generateConnectorName(source: DatabaseSource): String = {
-    s"${source.project}_${source.dbAlias}"
-  }
 }

gateway/Dockerfile

@@ -2,7 +2,7 @@ FROM golang:1.15.3-alpine3.12
 WORKDIR /build
 
 # Take the current space cloud version as a argument
-ARG SC_VERSION=0.20.1
+ARG SC_VERSION=0.21.0
 
 # Copy all the source files
 COPY . .
@@ -16,7 +16,7 @@ RUN GOOS=linux CGO_ENABLED=0 go build -a -ldflags '-s -w -extldflags "-static"'
 RUN echo $SC_VERSION && wget https://storage.googleapis.com/space-cloud/mission-control/mission-control-v$SC_VERSION.zip && unzip mission-control-v$SC_VERSION.zip
 
 FROM alpine:3.12
-ARG SC_VERSION=0.20.1
+ARG SC_VERSION=0.21.0
 
 RUN apk --no-cache add ca-certificates
 

gateway/config/config.go

@@ -303,19 +303,24 @@ type Service struct {
 
 // Endpoint holds the config of a endpoint
 type Endpoint struct {
-	Kind      EndpointKind     `json:"kind" yaml:"kind" mapstructure:"kind"`
-	Tmpl      TemplatingEngine `json:"template,omitempty" yaml:"template,omitempty" mapstructure:"template"`
-	ReqTmpl   string           `json:"requestTemplate" yaml:"requestTemplate" mapstructure:"requestTemplate"`
-	GraphTmpl string           `json:"graphTemplate" yaml:"graphTemplate" mapstructure:"graphTemplate"`
-	ResTmpl   string           `json:"responseTemplate" yaml:"responseTemplate" mapstructure:"responseTemplate"`
-	OpFormat  string           `json:"outputFormat,omitempty" yaml:"outputFormat,omitempty" mapstructure:"outputFormat"`
-	Token     string           `json:"token,omitempty" yaml:"token,omitempty" mapstructure:"token"`
-	Claims    string           `json:"claims,omitempty" yaml:"claims,omitempty" mapstructure:"claims"`
-	Method    string           `json:"method" yaml:"method" mapstructure:"method"`
-	Path      string           `json:"path" yaml:"path" mapstructure:"path"`
-	Rule      *Rule            `json:"rule,omitempty" yaml:"rule,omitempty" mapstructure:"rule"`
-	Headers   Headers          `json:"headers,omitempty" yaml:"headers,omitempty" mapstructure:"headers"`
-	Timeout   int              `json:"timeout,omitempty" yaml:"timeout,omitempty" mapstructure:"timeout"` // Timeout is in seconds
+	Kind EndpointKind     `json:"kind" yaml:"kind" mapstructure:"kind"`
+	Tmpl TemplatingEngine `json:"template,omitempty" yaml:"template,omitempty" mapstructure:"template"`
+	// ReqPayloadFormat specifies the payload format
+	// depending upon the payload format, the graphQL request that
+	// gets converted to http request will use that format as it's payload
+	// currently supported formats are application/json,multipart/form-data
+	ReqPayloadFormat string  `json:"requestPayloadFormat" yaml:"requestPayloadFormat" mapstructure:"requestPayloadFormat"`
+	ReqTmpl          string  `json:"requestTemplate" yaml:"requestTemplate" mapstructure:"requestTemplate"`
+	GraphTmpl        string  `json:"graphTemplate" yaml:"graphTemplate" mapstructure:"graphTemplate"`
+	ResTmpl          string  `json:"responseTemplate" yaml:"responseTemplate" mapstructure:"responseTemplate"`
+	OpFormat         string  `json:"outputFormat,omitempty" yaml:"outputFormat,omitempty" mapstructure:"outputFormat"`
+	Token            string  `json:"token,omitempty" yaml:"token,omitempty" mapstructure:"token"`
+	Claims           string  `json:"claims,omitempty" yaml:"claims,omitempty" mapstructure:"claims"`
+	Method           string  `json:"method" yaml:"method" mapstructure:"method"`
+	Path             string  `json:"path" yaml:"path" mapstructure:"path"`
+	Rule             *Rule   `json:"rule,omitempty" yaml:"rule,omitempty" mapstructure:"rule"`
+	Headers          Headers `json:"headers,omitempty" yaml:"headers,omitempty" mapstructure:"headers"`
+	Timeout          int     `json:"timeout,omitempty" yaml:"timeout,omitempty" mapstructure:"timeout"` // Timeout is in seconds
 }
 
 // EndpointKind describes the type of endpoint. Default value - internal
@@ -330,6 +335,12 @@ const (
 
 	// EndpointKindPrepared describes an endpoint on on Space Cloud GraphQL layer
 	EndpointKindPrepared EndpointKind = "prepared"
+
+	// EndpointRequestPayloadFormatJSON specifies json payload format for the request
+	EndpointRequestPayloadFormatJSON string = "json"
+
+	// EndpointRequestPayloadFormatFormData specifies multipart/form-data payload format for the request
+	EndpointRequestPayloadFormatFormData string = "form-data"
 )
 
 // TemplatingEngine describes the type of endpoint. Default value - go

gateway/go.mod

@@ -5,7 +5,7 @@ require (
 	github.com/Masterminds/goutils v1.1.0 // indirect
 	github.com/Masterminds/semver v1.5.0 // indirect
 	github.com/Masterminds/sprig v2.22.0+incompatible
-	github.com/aws/aws-sdk-go v1.33.14
+	github.com/aws/aws-sdk-go v1.34.28
 	github.com/cpuguy83/go-md2man/v2 v2.0.0 // indirect
 	github.com/denisenkom/go-mssqldb v0.0.0-20200428022330-06a60b6afbbc
 	github.com/dgrijalva/jwt-go v3.2.0+incompatible
@@ -37,16 +37,14 @@ require (
 	github.com/rs/cors v1.7.0
 	github.com/satori/go.uuid v1.2.0
 	github.com/segmentio/ksuid v1.0.3
-	github.com/spaceuptech/helpers v0.1.2
+	github.com/spaceuptech/helpers v0.2.1
 	github.com/spaceuptech/space-api-go v0.18.1
 	github.com/stretchr/objx v0.2.0 // indirect
-	github.com/stretchr/testify v1.6.1
-	github.com/tidwall/pretty v0.0.0-20180105212114-65a9db5fad51 // indirect
+	github.com/stretchr/testify v1.7.0
 	github.com/urfave/cli v1.22.2
-	github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c // indirect
 	github.com/xdg/stringprep v1.0.0 // indirect
 	go.etcd.io/bbolt v1.3.3
-	go.mongodb.org/mongo-driver v1.1.1
+	go.mongodb.org/mongo-driver v1.4.4
 	golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9
 	golang.org/x/net v0.0.0-20201006153459-a7d1128ccaa0
 	google.golang.org/api v0.18.0