Skip to content

Commit

Permalink
Merge pull request #1529 from spaceuptech/v0.21.0
Browse files Browse the repository at this point in the history 
v0.21.0
  • Loading branch information
@YourTechBud
YourTechBud committed Feb 19, 2021
2 parents 1f49871 + ed37cda commit 714ffb9
Show file tree
Hide file tree
Showing 182 changed files with 15,644 additions and 7,626 deletions.
1 change: 1 addition & 0 deletions dbevents/Dockerfile
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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 []
2 changes: 1 addition & 1 deletion dbevents/build.sbt
View file
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
name := "db-events-soruce"

version := "0.1.0"
version := "0.2.0"

scalaVersion := "2.13.1"

Expand Down
116 changes: 72 additions & 44 deletions dbevents/src/main/scala/com/spaceuptech/dbevents/database/Utils.scala
View file
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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,
Expand All @@ -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,
Expand Down Expand Up @@ -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")

Expand All @@ -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 = {
Expand Down Expand Up @@ -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)

Expand All @@ -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)

Expand Down Expand Up @@ -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}"
}
}
4 changes: 2 additions & 2 deletions gateway/Dockerfile
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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 . .
Expand All @@ -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

Expand Down
37 changes: 24 additions & 13 deletions gateway/config/config.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down
10 changes: 4 additions & 6 deletions gateway/go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
Loading

0 comments on commit 714ffb9

Please sign in to comment.