AMQP 1.0 (#91)

Completely rewrote message broker comm layer, removed support for RabbitMQ AMQP 0.9.1, added support for AMQP 1.0 - tested with RabbitMQ 4.2.0, AciveMQ Classic 6.1.8, ActiveMQ Artemis 2.44.0.

Fixed a few bugs in/around message handler.

Improvements in TF deployment.

Removed k8s POC and Amazon Keyspaces deployment, they are now in https://github.com/capillariesio/capillaries-experimental.

Added experimental CapiMQ message broker.

Author: kleineshertz

.github/workflows/go.yml

@@ -53,6 +53,19 @@ jobs:
                 exit 1
             fi
 
+    - name: pkg/capimq_message_broker test coverage threshold check
+      env:
+          TESTCOVERAGE_THRESHOLD: 82.4
+      run: |
+            go test -v ./pkg/capimq_message_broker/... -coverprofile coverage.out -covermode count
+            totalCoverage=`go tool cover -func=coverage.out | grep total | grep -Eo '[0-9]+\.[0-9]+'`
+            if (( $(echo "$totalCoverage $TESTCOVERAGE_THRESHOLD" | awk '{print ($1 >= $2)}') )); then
+                echo -e "\033[32mOK: $totalCoverage >= $TESTCOVERAGE_THRESHOLD\033[0m"
+            else
+                echo -e "\033[31mFAILED: $totalCoverage < $TESTCOVERAGE_THRESHOLD. Cover more with unit tests or adjust threshold to a lower value.\033[0m"
+                exit 1
+            fi
+
     - name: pkg/cql test coverage threshold check
       env:
           TESTCOVERAGE_THRESHOLD: 88.1
@@ -81,7 +94,7 @@ jobs:
 
     - name: pkg/custom/tag_and_denormalize test coverage threshold check
       env:
-          TESTCOVERAGE_THRESHOLD: 76.1
+          TESTCOVERAGE_THRESHOLD: 78.5
       run: |
             go test -v ./pkg/custom/tag_and_denormalize/... -coverprofile coverage.out -covermode count
             totalCoverage=`go tool cover -func=coverage.out | grep total | grep -Eo '[0-9]+\.[0-9]+'`
@@ -94,7 +107,7 @@ jobs:
 
     - name: pkg/dpc test coverage threshold check
       env:
-          TESTCOVERAGE_THRESHOLD: 85.7
+          TESTCOVERAGE_THRESHOLD: 80.0
       run: |
             go test -v ./pkg/dpc/... -coverprofile coverage.out -covermode count
             totalCoverage=`go tool cover -func=coverage.out | grep total | grep -Eo '[0-9]+\.[0-9]+'`
@@ -120,7 +133,7 @@ jobs:
 
     - name: pkg/sc test coverage threshold check
       env:
-          TESTCOVERAGE_THRESHOLD: 89.6
+          TESTCOVERAGE_THRESHOLD: 89.5
       run: |
             go test -v ./pkg/sc/... -coverprofile coverage.out -covermode count
             totalCoverage=`go tool cover -func=coverage.out | grep total | grep -Eo '[0-9]+\.[0-9]+'`

.golangci.yml

@@ -133,4 +133,12 @@ linters:
         linters: [ revive ]
       - path: pkg\/proc\/proc_file_creator\.go
         text: "unchecked-type-assertion: type cast result is unchecked in [^ ]+FileRecordHeapItem"
-        linters: [ revive ]
+        linters: [ revive ]
+      - path: pkg\/api\/message_handler\.go
+        text: function ProcessDataBatchMsg has cyclomatic complexity
+        linters: [ revive ]
+      # - path: .+\.go
+      #   text: "unused-parameter: parameter '_"
+      #   linters: [ revive ]
+
+        

.vscode/launch.json

@@ -107,6 +107,16 @@
             "program": "${workspaceFolder}/pkg/exe/webapi/capiwebapi.go",
             "args": []
         },
+        {
+            "name": "CapiMqBroker",
+            "type": "go",
+            "request": "launch",
+            "mode": "debug",
+            "envFile": "${env:HOME}/vscode-daemon.env",
+            "cwd":"${workspaceFolder}/pkg/exe/mq",
+            "program": "${workspaceFolder}/pkg/exe/mq/capimq.go",
+            "args": []
+        },
         {
             "name": "capiparquet diff",
             "type": "go",

README.md

@@ -18,7 +18,7 @@ This is a GitHub readme. More details and a blog at https://capillaries.io.
 | Data filtering | SQL queries, custom code | [Go expressions](doc/glossary.md#go-expressions) (scalability, maintainability) |
 | Data transform | SQL expressions, custom code | [Go expressions](doc/glossary.md#go-expressions), Python [formulas](doc/glossary.md#py_calc-processor) (parallel execution, maintainability) |
 | Intermediate data storage | Files, relational databases | on-the-fly-created Cassandra [keyspaces](doc/glossary.md#keyspace) and [tables](doc/glossary.md#table) (scalability, maintainability) |
-| Workflow execution | Shell scripts, custom code, workflow frameworks | RabbitMQ as scheduler, workflow status stored in Cassandra (parallel execution, fault tolerance, incremental computing) |
+| Workflow execution | Shell scripts, custom code, workflow frameworks | Message queue as scheduler, workflow status stored in Cassandra (parallel execution, fault tolerance, incremental computing) |
 | Workflow monitoring and interaction | Custom solutions | Capillaries [UI](ui/README.md), [Toolbelt](doc/glossary.md#toolbelt) utility, [API](doc/api.md), [Web API](doc/glossary.md#webapi) (transparency, operator validation support) |
 | Workflow management | Shell scripts, custom code | Capillaries configuration: [script file](doc/glossary.md#script) with [DAG](doc/glossary.md#dag), Python [formulas](doc/glossary.md#py_calc-processor) |
 
@@ -65,7 +65,7 @@ Log messages generated by:
 - Capillaries [Daemon](./doc/glossary.md#daemon)
 - Capillaries [WebAPI](./doc/glossary.md#webapi)
 - Capillaries [UI](./doc/glossary.md#capillaries-ui)
-- RabbitMQ
+- Message queue
 - Cassandra with Prometheus jmx-exporter
 - Prometheus
 are collected by fluentd and saved in /tmp/capi_log.
@@ -91,11 +91,11 @@ For more details about getting started, see [Getting started](doc/started.md).
 
 #### Container-based deployments
 
-Capillaries binaries are intended to be container-friendly. Check out the `docker-compose.yml` and [Kubernetes deployment POC](./deploy/k8s/README.md), these test projects may be a good starting point for creating your full-scale container-based deployment.
+Capillaries binaries are intended to be container-friendly. Check out the `docker-compose.yml`, it may be a good starting point for creating your full-scale container-based deployment.
 
 #### VM-based deployment
 
-See [Terraform script](./deploy/tf/cassandra_cluster/README.md) that creates Capillaries deployment in AWS.
+See [Terraform script](./deploy/README.md) that creates Capillaries deployment in AWS.
 
 ## Capillaries in depth
 

binaries_build.sh

@@ -1,11 +1,15 @@
 #!/bin/bash
 
-DIR_BUILD_LINUX_AMD64=./build/linux/amd64
-DIR_BUILD_LINUX_ARM64=./build/linux/arm64
+DIR_BUILD_LINUX=./build/linux
+DIR_BUILD_LINUX_AMD64=$DIR_BUILD_LINUX/amd64
+DIR_BUILD_LINUX_ARM64=$DIR_BUILD_LINUX/arm64
 DIR_BUILD_CA=./build/ca
 DIR_BUILD_WEBUI=./build/webui
 
-rm -fR ./build
+rm -fR $DIR_BUILD_CA
+rm -fR $DIR_BUILD_LINUX
+rm -fR $DIR_BUILD_WEBUI
+
 mkdir -p $DIR_BUILD_LINUX_AMD64
 mkdir -p $DIR_BUILD_LINUX_ARM64
 mkdir -p $DIR_BUILD_CA
@@ -45,6 +49,12 @@ gzip -f -k $DIR_BUILD_LINUX_AMD64/capitoolbelt
 GOOS=linux GOARCH=amd64 go build -o $DIR_BUILD_LINUX_AMD64/capiparquet -ldflags="-s -w -X main.version=$(git describe --tags --always)" ./test/code/parquet/capiparquet.go
 gzip -f -k $DIR_BUILD_LINUX_AMD64/capiparquet
 
+GOOS=linux GOARCH=arm64 go build -o $DIR_BUILD_LINUX_AMD64/capimq -ldflags="-s -w -X main.version=$(git describe --tags --always)" ./pkg/exe/mq/capimq.go
+cp ./pkg/exe/mq/capimq.json $DIR_BUILD_LINUX_AMD64
+gzip -f -k $DIR_BUILD_LINUX_AMD64/capimq
+
+
+
 echo "Building "$DIR_BUILD_LINUX_ARM64
 
 GOOS=linux GOARCH=arm64 go build -o $DIR_BUILD_LINUX_ARM64/capidaemon -ldflags="-s -w -X main.version=$(git describe --tags --always)" ./pkg/exe/daemon/capidaemon.go
@@ -61,3 +71,7 @@ gzip -f -k $DIR_BUILD_LINUX_ARM64/capitoolbelt
 
 GOOS=linux GOARCH=arm64 go build -o $DIR_BUILD_LINUX_ARM64/capiparquet -ldflags="-s -w -X main.version=$(git describe --tags --always)" ./test/code/parquet/capiparquet.go
 gzip -f -k $DIR_BUILD_LINUX_ARM64/capiparquet
+
+GOOS=linux GOARCH=arm64 go build -o $DIR_BUILD_LINUX_ARM64/capimq -ldflags="-s -w -X main.version=$(git describe --tags --always)" ./pkg/exe/mq/capimq.go
+cp ./pkg/exe/mq/capimq.json $DIR_BUILD_LINUX_ARM64
+gzip -f -k $DIR_BUILD_LINUX_ARM64/capimq

binaries_upload.sh

@@ -43,4 +43,6 @@ if [ "$1" = "" ]; then
 fi
 
 echo "Copying in files to "$1
-aws s3 cp ./build/ $1/ --recursive --include "*"
+aws s3 cp ./build/ca $1/ca/ --recursive --include "*"
+aws s3 cp ./build/linux $1/linux/ --recursive --include "*"
+aws s3 cp ./build/webui $1/webui/ --recursive --include "*"

deploy/tf/cassandra_cluster/README.md deploy/README.mddeploy/tf/cassandra_cluster/README.md renamed to deploy/README.md

@@ -20,11 +20,11 @@ TF variables: `ssh_keypair_name`
 
 ## S3 bucket
 
-Make sure you have an S3 bucket as explained [Capillaries S3 doc](../../../doc/s3.md). Most likely, you will need to set up a user [UserAccessCapillariesTestbucket](../../../doc/s3.md#iam-user-useraccesscapillariestestbucket), otherwise you will not be able to upload source data and configuration files to the bucket from your devops machine. Make sure this user has full access to the bucket as the doc says.
+Make sure you have an S3 bucket as explained [Capillaries S3 doc](../doc/s3.md). Most likely, you will need to set up a user [UserAccessCapillariesTestbucket](../doc/s3.md#iam-user-useraccesscapillariestestbucket), otherwise you will not be able to upload source data and configuration files to the bucket from your devops machine. Make sure this user has full access to the bucket as the doc says.
 
 EC2 instances of this deployment do NOT need UserAccessCapillariesTestbucket credentials, the instances use AssumeRole mechanism to access S3 bucket.
 
-This example writes [Webapi](../../../doc/glossary.md#webapi) and [Daemon](../../../doc/glossary.md#daemon) log files to the `log` directory in your bucket.
+This example writes [Webapi](../doc/glossary.md#webapi) and [Daemon](../doc/glossary.md#daemon) log files to the `log` directory in your bucket.
 
 TF variables: `s3_log_url`
 
@@ -94,7 +94,7 @@ export CAPILLARIES_AWS_TESTBUCKET=capillaries-testbucket
 export BASTION_IP=your_bastion_ip_address
 export EXTERNAL_WEBAPI_PORT=6544
 
-# Remember UserAccessCapillariesTestbucket introduced in ../../../doc/s3.md ?
+# Remember UserAccessCapillariesTestbucket introduced in ../doc/s3.md ?
 source ~/UserAccessCapillariesTestbucket.rc
 
 # Full test cycle

…ploy/tf/cassandra_cluster/bastion.sh.tpl deploy/bastion.sh.tpldeploy/tf/cassandra_cluster/bastion.sh.tpl renamed to deploy/bastion.sh.tpl deploy/tf/cassandra_cluster/bastion.sh.tpl renamed to deploy/bastion.sh.tpl

@@ -0,0 +1,46 @@
+resource "aws_network_interface" "bastion_internal_ip" {
+  subnet_id   = aws_subnet.public_subnet.id
+  private_ips = [var.internal_bastion_ip]
+  security_groups = [aws_security_group.capillaries_securitygroup_bastion.id]
+
+  tags = {
+    Name = "capillaries_bastion_internal_ip"
+  }
+}
+
+resource "aws_eip_association" "bastion_public_ip_association" {
+  instance_id   = aws_instance.bastion.id
+  allocation_id = aws_eip.bastion_public_ip.id
+}
+
+# Make sure it matched the list of expected variables in bastion.sh (bastion_provisioner_vars add a bit too)
+locals {
+  bastion_provisioner_static_vars = "WEBAPI_GOMEMLIMIT_GB=${local.webapi_gomemlimit_gb} WEBAPI_GOGC=${var.webapi_gogc} AWSREGION=${var.awsregion} SSH_USER=${var.ssh_user} OS_ARCH=${local.os_arch} CAPILLARIES_RELEASE_URL=${var.capillaries_release_url} EXTERNAL_WEBAPI_PORT=${var.external_webapi_port} INTERNAL_WEBAPI_PORT=${var.internal_webapi_port} EXTERNAL_RABBITMQ_CONSOLE_PORT=${var.external_rabbitmq_console_port} EXTERNAL_ACTIVEMQ_CONSOLE_PORT=${var.external_activemq_console_port} EXTERNAL_PROMETHEUS_CONSOLE_PORT=${var.external_prometheus_console_port} BASTION_ALLOWED_IPS=${var.BASTION_ALLOWED_IPS} S3_LOG_URL=${var.s3_log_url} CASSANDRA_HOSTS=${local.cassandra_hosts} CASSANDRA_PORT=${var.cassandra_port} CASSANDRA_USERNAME=${var.cassandra_username} CASSANDRA_PASSWORD=${var.cassandra_password} MQ_TYPE=${var.mq_type} CAPIMQ_CLIENT_URL=${local.capimq_url} AMQP10_URL=${local.activemq_url} AMQP10_ADDRESS=${var.amqp10_flavor_address_map[var.amqp10_server_flavor]} AMQP10_USER_NAME=${var.amqp10_user_name} AMQP10_USER_PASS=${var.amqp10_user_pass} AMQP10_ADMIN_NAME=${var.amqp10_admin_name} AMQP10_ADMIN_PASS=${var.amqp10_admin_pass} AMQP10_SERVER_FLAVOR=${var.amqp10_server_flavor} PROMETHEUS_NODE_EXPORTER_FILENAME=${local.prometheus_node_exporter_filename} PROMETHEUS_SERVER_FILENAME=${local.prometheus_server_filename} PROMETHEUS_NODE_TARGETS=${local.prometheus_node_targets} PROMETHEUS_JMX_TARGETS=${local.prometheus_jmx_targets} PROMETHEUS_GO_TARGETS=${local.prometheus_go_targets} RABBITMQ_ERLANG_FILENAME=${local.rabbitmq_erlang_filename} RABBITMQ_SERVER_FILENAME=${local.rabbitmq_server_filename} ACTIVEMQ_CLASSIC_SERVER_FILENAME=${local.activemq_classic_server_filename} ACTIVEMQ_ARTEMIS_SERVER_FILENAME=${local.activemq_artemis_server_filename} INTERNAL_CAPIMQ_BROKER_PORT=${var.internal_capimq_broker_port} EXTERNAL_CAPIMQ_BROKER_PORT=${var.external_capimq_broker_port} CAPIMQ_BROKER_MAX_MESSAGES=${var.capimq_broker_max_messages} CAPIMQ_BROKER_RETURNED_DELIVERY_DELAY=${var.capimq_broker_returned_delivery_delay} CAPIMQ_BROKER_DEAD_AFTER_NO_HEARTBEAT_TIMEOUT=${var.capimq_broker_dead_after_no_heartbeat_timeout} WEBAPI_PROMETHEUS_EXPORTER_PORT=${var.webapi_prometheus_exporter_port} CAPIMQ_PROMETHEUS_EXPORTER_PORT=${var.capimq_prometheus_exporter_port}"
+}
+
+resource "aws_instance" "bastion" {
+  instance_type          = var.bastion_instance_type
+  ami                    = var.bastion_ami_name
+  key_name               = var.ssh_keypair_name
+
+  network_interface {
+    network_interface_id = aws_network_interface.bastion_internal_ip.id
+    device_index         = 0
+  }
+  
+  # Bastion needs to assume this role to access S3 bucket to get cloud-init bastion.sh
+  iam_instance_profile = aws_iam_instance_profile.capillaries_instance_profile.name
+
+  user_data              = templatefile("./bastion.sh.tpl", {
+    os_arch                                = local.os_arch
+    ssh_user                               = var.ssh_user
+    capillaries_instance_profile           = aws_iam_instance_profile.capillaries_instance_profile.name
+    # Used in bastion.sh.tpl
+    bastion_provisioner_vars               = join(" ", [local.bastion_provisioner_static_vars], ["BASTION_EXTERNAL_IP_ADDRESS=${aws_eip.bastion_public_ip.public_ip}"])
+    capillaries_tf_deploy_temp_bucket_name = var.capillaries_tf_deploy_temp_bucket_name
+  })
+  
+  tags = {
+    Name = "capillaries_bastion"
+  }
+}