updated instruction for labs

labs/data_generator/generator.py

@@ -55,7 +55,8 @@ class KinesisProducer:
                     Data=payload,
                     PartitionKey=key
                 )
-                logger.info('Msg with trans_id={} sent to shard {} seq no {}'.format(tran_id, response["ShardId"], response["SequenceNumber"]))
+                logger.info('Msg with trans_id={} sent to shard {} seq no {}'.format(tran_id, response["ShardId"],
+                                                                                     response["SequenceNumber"]))
                 return response
 
             except Exception as e:
@@ -88,7 +89,7 @@ def prepare_event(event):
     return msg_formatted, msg_key
 
 
-def produce_data(kinesis_data_stream, messages_per_sec, input_file):
+def produce_data(kinesis_data_stream, messages_per_sec, input_file, single_run):
     """
     Main method for producing
     :param kinesis_data_stream: param from cmdline name of KDS
@@ -118,6 +119,8 @@ def produce_data(kinesis_data_stream, messages_per_sec, input_file):
             event, key = prepare_event(row)
             kp.produce(event, key, kinesis_data_stream)
 
+        if single_run:
+            break
         replay_cnt += 1
 
 
@@ -128,16 +131,20 @@ if __name__ == "__main__":
     parser.add_argument('-k', '--kinesis_ds', dest='kinesis_ds', required=True)
     parser.add_argument('-i', '--input_file', dest='input_file', required=False)
     parser.add_argument('-s', '--messages_per_sec', dest='mps', type=int, default=-1, required=False)
+    parser.add_argument('-r', '--single-run', dest='singel_run', action='store_true', required=False, default=False)
 
     args, unknown = parser.parse_known_args()
     config = configparser.ConfigParser()
 
     kinesis_data_stream = args.kinesis_ds
     messages_per_sec = int(args.mps)
+
+    single_run = args.singel_run if hasattr(args, 'singel_run') else False
+
     if args.input_file:
         input_file = args.input_file
     else:
         main_path = os.path.abspath(os.path.dirname(__file__))
         input_file = os.path.join(main_path, DEFAULT_DATA_FILE)
 
-    produce_data(kinesis_data_stream, messages_per_sec, input_file)
+    produce_data(kinesis_data_stream, messages_per_sec, input_file, single_run)

labs/labs_1.md

@@ -1,14 +0,0 @@
-
-Utwórz nowy folder w swoim repozytorium kodu ./cdp-terraform
-skopiuj pliki
-
-
-
-
-1. Stwórz S3 bucket
-- sprawdź definicję https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/s3_bucket
-
-
-2. Stwórz Kinesis Data Stream. Przyjmij do konfiguracji następujące parametry
-- nazwa : <env = dev>_<inicjały_studenta>_<nr_indeksu> np dev_jk_12345
-- liczba shardów - 1

labs/labs_preparation.md

@@ -7,7 +7,7 @@
 1. Wymagania wstępne - środowiska (rekomendowane **PyCharm + Anacoda**)
 * PyCharm - https://www.jetbrains.com/pycharm/download/
 * Anaconda - https://www.anaconda.com/products/individual#Downloads
-  - nowe środowisko Python 3.8  
+  - nowe środowisko Python 3.9 
     Windows users : użyj Anaconda Prompt)  
     Linux / MacOs bash / zsh etc..
     ```

labs/requirements.txt

@@ -1,4 +1,4 @@
-awscli==1.19.33
-boto3==1.17.33
-configparser==5.0.2
-awswrangler==2.7.0
+awscli==1.32.91
+boto3==1.34.89
+configparser==7.0.0
+awswrangler==3.7.3

labs/setup_c9.sh

@@ -0,0 +1,12 @@
+#!/bin/bash
+sudo yum update -y
+
+wget https://releases.hashicorp.com/terraform/1.8.1/terraform_1.8.1_linux_amd64.zip -P ~/
+unzip ~/terraform_1.8.1_linux_amd64.zip -d ~/.
+sudo mv ~/terraform /usr/local/bin
+
+pip install -r requirements.txt
+
+echo "alias python='python3'
+alias tf='terraform'
+alias c='clear'" >> ~/.bashrc

labs/terraform/main.tf

@@ -1,7 +1,7 @@
 locals {
   common_tags = {
-    Purpose = "UAM Cloud Data Processing"
-    Environment = "DEV"
-    Owner = var.student_full_name
+    purpose = "UAM Cloud Data Processing"
+    environment = "DEV"
+    owner = var.student_full_name
   }
 }

labs/terraform/provider.tf

@@ -2,7 +2,7 @@ terraform {
   required_providers {
     aws = {
       source = "hashicorp/aws"
-      version = "~> 3.27"
+      version = "~> 5.0"
     }
   }
 }

labs/terraform/solutions/glue.tf

@@ -1,17 +0,0 @@
-resource "aws_glue_catalog_database" "datalake_db_raw_zone" {
-  name = "datalake_${var.environment}_${var.account_number}_${var.student_initials}_${var.student_index_no}"
-}
-
-
-resource "aws_glue_crawler" "glue_crawler_raw_zone" {
-  database_name = aws_glue_catalog_database.datalake_db_raw_zone.name
-  name = "gc-raw-${var.environment}-${var.account_number}-${var.student_initials}-${var.student_index_no}"
-  role = aws_iam_role.glue_crawler_role.arn
-  table_prefix = "crawler_"
-
-  s3_target {
-    path = "s3://${aws_s3_bucket.main_dl_bucket.bucket}/raw-zone/stockdata/"
-  }
-
-  tags = merge(local.common_tags, )
-}

labs/terraform/solutions/iam.tf

@@ -1,171 +0,0 @@
-resource "aws_iam_role" "firehose_stream_role" {
-  name = "firehose-role-${var.environment}-${var.account_number}-${var.student_initials}-${var.student_index_no}"
-
-  assume_role_policy = <<EOF
-{
-   "Version": "2012-10-17",
-   "Statement": [
-     {
-       "Action": "sts:AssumeRole",
-       "Principal": {
-         "Service": "firehose.amazonaws.com"
-       },
-       "Effect": "Allow",
-       "Sid": ""
-     }
-   ]
-}
-EOF
-}
-
-resource "aws_iam_role_policy" "firehose_stream_policy" {
-  name = "firehose-stream-policy-${var.environment}-${var.account_number}-${var.student_initials}-${var.student_index_no}"
-  role = aws_iam_role.firehose_stream_role.id
-
-  policy = <<EOF
-{
-   "Version": "2012-10-17",
-   "Statement": [
-      {
-           "Effect": "Allow",
-           "Action": "kinesis:*",
-           "Resource": "*"
-       },
-      {
-           "Effect": "Allow",
-           "Action": [
-               "s3:AbortMultipartUpload",
-               "s3:GetBucketLocation",
-               "s3:GetObject",
-               "s3:ListBucket",
-               "s3:ListBucketMultipartUploads",
-               "s3:PutObject"
-           ],
-           "Resource": [
-             "${aws_s3_bucket.main_dl_bucket.arn}",
-             "${aws_s3_bucket.main_dl_bucket.arn}/*"
-           ]
-       },
-       {
-           "Sid": "",
-           "Effect": "Allow",
-           "Action": [
-               "logs:PutLogEvents"
-           ],
-           "Resource": [
-               "arn:aws:logs:${var.region}:${var.account_number}:log-group:/aws/kinesisfirehose/*"
-           ]
-       }
-   ]
-}
-EOF
-}
-
-
-// Role & policies for Glue Crawler
-resource "aws_iam_role" "glue_crawler_role" {
-  name = "crawler-role-${var.environment}-${var.account_number}-${var.student_initials}-${var.student_index_no}"
-
-  assume_role_policy = <<EOF
-{
-    "Version": "2012-10-17",
-    "Statement": [
-      {
-        "Effect": "Allow",
-        "Principal": {
-          "Service": "glue.amazonaws.com"
-        },
-        "Action": "sts:AssumeRole"
-      }
-    ]
-}
-EOF
-}
-
-
-data "aws_iam_policy" "glue_service_policy" {
-  arn = "arn:aws:iam::aws:policy/service-role/AWSGlueServiceRole"
-}
-
-resource "aws_iam_role_policy" "glue_crawler_user_bucket_policy" {
-  name = "user-bucket-policy-${var.environment}-${var.account_number}-${var.student_initials}-${var.student_index_no}"
-  role = aws_iam_role.glue_crawler_role.id
-
-  policy = <<EOF
-{
-   "Version": "2012-10-17",
-   "Statement": [
-    {
-        "Effect": "Allow",
-        "Action": [
-            "s3:GetObject",
-            "s3:PutObject"
-        ],
-        "Resource": [
-             "${aws_s3_bucket.main_dl_bucket.arn}*"
-        ]
-    }
-   ]
-}
-EOF
-}
-
-resource "aws_iam_policy_attachment" "crawler_attach_managed_policy" {
-  name = "crawler-managed-service-${var.environment}-${var.account_number}-${var.student_initials}-${var.student_index_no}"
-  roles = [
-    aws_iam_role.glue_crawler_role.name]
-  policy_arn = data.aws_iam_policy.glue_service_policy.arn
-}
-
-
-// Role and policies for Lambda
-resource "aws_iam_role" "lambda_basic_role" {
-  name = "lambda-basic-role-${var.environment}-${var.account_number}-${var.student_initials}-${var.student_index_no}"
-  tags = merge(local.common_tags, )
-
-  assume_role_policy = <<EOF
-{
- "Version": "2012-10-17",
-  "Statement": [
-    {
-      "Action": "sts:AssumeRole",
-      "Principal": {
-        "Service": "lambda.amazonaws.com"
-      },
-      "Effect": "Allow",
-      "Sid": ""
-    }
-  ]
-}
-EOF
-}
-
-resource "aws_iam_role_policy" "lambda_basic_policy" {
-  name = "lambda-basic-policy-${var.environment}-${var.account_number}-${var.student_initials}-${var.student_index_no}"
-  role = aws_iam_role.lambda_basic_role.id
-
-  policy = <<EOF
-{
- "Version": "2012-10-17",
- "Statement": [
-   {
-     "Action": [
-       "logs:CreateLogGroup",
-       "logs:CreateLogStream",
-       "logs:PutLogEvents"
-     ],
-     "Effect": "Allow",
-     "Resource": "*"
-   },
-   {
-     "Effect": "Allow",
-     "Action": "s3:*",
-     "Resource": [
-             "${aws_s3_bucket.main_dl_bucket.arn}",
-             "${aws_s3_bucket.main_dl_bucket.arn}/*"]
-   }
- ]
-}
-EOF
-}
-

labs/terraform/solutions/kinesis_ds.tf

@@ -1,14 +0,0 @@
-resource "aws_kinesis_stream" "cryptostock_stream" {
-  name = "cryptostock-${var.environment}-${var.account_number}-${var.student_initials}-${var.student_index_no}"
-  shard_count = 1
-  enforce_consumer_deletion = true
-
-  shard_level_metrics = [
-    "IncomingBytes",
-    "OutgoingBytes",
-    "IncomingRecords",
-    "OutgoingRecords"
-  ]
-
-  tags = merge(local.common_tags, )
-}

labs/terraform/solutions/kinesis_fh.tf

@@ -1,18 +0,0 @@
-resource "aws_kinesis_firehose_delivery_stream" "stock_delivery_stream" {
-  name = "firehose-${var.environment}-${var.account_number}-${var.student_initials}-${var.student_index_no}"
-  destination = "extended_s3"
-
-  kinesis_source_configuration {
-    kinesis_stream_arn = aws_kinesis_stream.cryptostock_stream.arn
-    role_arn = aws_iam_role.firehose_stream_role.arn
-  }
-
-  extended_s3_configuration {
-    role_arn = aws_iam_role.firehose_stream_role.arn
-    bucket_arn = aws_s3_bucket.main_dl_bucket.arn
-    buffer_size = 1
-    buffer_interval = 60
-    prefix = "raw-zone/stockdata/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/hour=!{timestamp:HH}/"
-    error_output_prefix = "${         "raw-zone/stockdata_errors/!{firehose:error-output-type}/year=!{timestamp:yyyy}"}${        "/month=!{timestamp:MM}/day=!{timestamp:dd}/hour=!{timestamp:HH}"}/"
-  }
-}

labs/terraform/solutions/lambda.tf

@@ -1,50 +0,0 @@
-resource "aws_lambda_layer_version" "aws_wrangler" {
-  filename = "../lambda/awswrangler-layer-2.7.0-py3.8.zip"
-  layer_name = "aws_wrangler_${var.environment}_${var.account_number}_${var.student_initials}_${var.student_index_no}"
-  source_code_hash = "${filebase64sha256("../lambda/awswrangler-layer-2.7.0-py3.8.zip")}"
-  compatible_runtimes = [
-    "python3.8"]
-}
-
-
-resource "aws_lambda_function" "etl_post_processing" {
-
-  function_name = "etl-post-processing-${var.environment}-${var.account_number}-${var.student_initials}-${var.student_index_no}"
-  filename = "../lambda/lambda_definition.zip"
-  handler = "lambda_definition.etl_function"
-  runtime = "python3.8"
-  role = aws_iam_role.lambda_basic_role.arn
-  timeout = 300
-  memory_size = 512
-  source_code_hash = filebase64sha256("../lambda/lambda_definition.zip")
-  layers = [
-    "${aws_lambda_layer_version.aws_wrangler.arn}"]
-
-}
-
-
-resource "aws_lambda_permission" "allow_bucket" {
-  statement_id = "AllowExecutionFromS3Bucket"
-  action = "lambda:InvokeFunction"
-  function_name = aws_lambda_function.etl_post_processing.arn
-  principal = "s3.amazonaws.com"
-  source_arn = aws_s3_bucket.main_dl_bucket.arn
-}
-
-
-resource "aws_s3_bucket_notification" "trigger_etl_lambda" {
-  bucket = aws_s3_bucket.main_dl_bucket.id
-
-  lambda_function {
-    lambda_function_arn = aws_lambda_function.etl_post_processing.arn
-    events = [
-      "s3:ObjectCreated:*"]
-    filter_prefix = "raw-zone/"
-  }
-
-  depends_on = [
-    aws_lambda_permission.allow_bucket]
-}
-
-
-

labs/terraform/solutions/main.tf

@@ -1,7 +0,0 @@
-locals {
-  common_tags = {
-    Purpose = "UAM Cloud Data Processing"
-    Environment = "DEV"
-    Owner = var.student_full_name
-  }
-}

labs/terraform/solutions/queries.sql

@@ -1,75 +0,0 @@
--- lab 3.3
-
-WITH CTE AS
-(
-
- SELECT date_format(from_unixtime(transaction_ts),'%Y-%m-%dT%H') as HourlyBucket,
-  RANK() OVER(PARTITION BY  date_format(from_unixtime(transaction_ts),'%Y-%m-%dT%H'), symbol ,type ORDER BY dollar_amount DESC) as rnk, *
- FROM "datalake_dev_100603781557_jk_12345"."crawler_stockdata"
-
-)
-select *
-from CTE
-where rnk=1
-order by 1, 4, 8
-
--- LAB 4.2
-
-CREATE EXTERNAL TABLE processed_stockdata(
-  transaction_date timestamp,
-  price double,
-  amount double,
-  dollar_amount double,
-  type string,
-  trans_id bigint)
-PARTITIONED BY (
-  symbol string,
-  year integer,
-  month integer,
-  day integer,
-  hour integer
-  )
-ROW FORMAT SERDE
-  'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
-STORED AS INPUTFORMAT
-  'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat'
-OUTPUTFORMAT
-  'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
-LOCATION
-  's3://datalake-dev-100603781557-jk-12345/processed-zone/stockdata/'
-
-
-MSCK REPAIR TABLE processed_stockdata;
-
--- LAB 5
-
---          .----------.   .----------.   .----------.
---          |  SOURCE  |   |  INSERT  |   |  DESTIN. |
--- Source-->|  STREAM  |-->| & SELECT |-->|  STREAM  |-->Destination
---          |          |   |  (PUMP)  |   |          |
---          '----------'   '----------'   '----------'
-
-
-CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM"
-("symbol" VARCHAR(10), "type" VARCHAR(10), "trans_id" BIGINT,
-    "dollar_amount" DOUBLE, "AvgLast30seconds" DOUBLE, "CntLast30seconds"  INT,
-    "SumLast30rows" DOUBLE, "CntLast30rows"  INT, "max_tran_id" BIGINT );
-
-CREATE OR REPLACE PUMP "STREAM_PUMP" AS INSERT INTO "DESTINATION_SQL_STREAM"
-SELECT STREAM "symbol", "type", "trans_id", "dollar_amount", "AvgLast30seconds", "CntLast30seconds"
- , "SumLast30rows", "CntLast30rows", "max_tran_id"
-FROM (
-
-    SELECT STREAM "symbol", "type", "trans_id", "dollar_amount",
-        AVG("dollar_amount") OVER LAST_30_SECS AS "AvgLast30seconds",
-        COUNT(*)  OVER LAST_30_SECS AS "CntLast30seconds",
-        SUM("dollar_amount") OVER LAST_30_ROWS AS "SumLast30rows",
-        COUNT(*)  OVER LAST_30_ROWS AS "CntLast30rows",
-        MAX("trans_id") OVER LAST_30_ROWS AS "max_tran_id"
-    FROM "SOURCE_SQL_STREAM_001"
-    WHERE "symbol" = 'BTC_USD'
-    WINDOW
-        LAST_30_SECS AS (PARTITION BY "symbol", "type" RANGE INTERVAL '30' SECOND PRECEDING),
-        LAST_30_ROWS AS (PARTITION BY "symbol", "type" ROWS 30 PRECEDING)
-)
-WHERE "dollar_amount" > 4 * ("AvgLast30seconds");

labs/terraform/solutions/s3.tf

@@ -1,6 +0,0 @@
-resource "aws_s3_bucket" "main_dl_bucket" {
-  bucket = "datalake-${var.environment}-${var.account_number}-${var.student_initials}-${var.student_index_no}"
-  force_destroy = true
-
-  tags = merge(local.common_tags, )
-}

labs/terraform/starter_files/main.tf

@@ -1,7 +1,7 @@
 locals {
   common_tags = {
-    Purpose = "UAM Cloud Data Processing"
-    Environment = "DEV"
-    Owner = var.student_full_name
+    purpose = "UAM Cloud Data Processing"
+    environment = "DEV"
+    owner = var.student_full_name
   }
 }

labs/terraform/starter_files/provider.tf

@@ -2,7 +2,7 @@ terraform {
   required_providers {
     aws = {
       source = "hashicorp/aws"
-      version = "~> 3.27"
+      version = "~> 5.0"
     }
   }
 }

labs/terraform/starter_files/terraform.tfvars

@@ -1,4 +1,5 @@
-account_number=920628590621
+account_number=XXXXX
 student_initials="jk"
 student_full_name="Jakub Kasprzak"
 student_index_no = "12345"
+lab_role_arn = "arn:aws:iam::XXXXX:role/LabRole"

labs/terraform/starter_files/variables.tf

@@ -29,3 +29,9 @@ variable "student_index_no" {
   description = "Index no"
   type = string
 }
+
+variable "lab_role_arn" {
+  description = "the role we use for all labs, dont use a single role for everything! it is an anti-pattern!!!!"
+  type = string
+
+}

labs/terraform/terraform.tfvars

@@ -1,4 +1,5 @@
-account_number=920628590621
+account_number=XXXXX
 student_initials="jk"
 student_full_name="Jakub Kasprzak"
 student_index_no = "12345"
+lab_role_arn = "arn:aws:iam::XXXXX:role/LabRole"

labs/terraform/variables.tf

@@ -29,3 +29,9 @@ variable "student_index_no" {
   description = "Index no"
   type = string
 }
+
+variable "lab_role_arn" {
+  description = "the role we use for all labs, dont use a single role for everything! it is an anti-pattern!!!!"
+  type = string
+
+}

testing-stack/docker-compose.yml

@@ -1,100 +0,0 @@
-version: '3.7'
-
-services:
-  zookeeper:
-    image: confluentinc/cp-zookeeper:5.2.2
-    ports:
-      - 2181:2181
-    hostname: zookeeper
-    environment:
-      ZOOKEEPER_CLIENT_PORT: 2181
-    networks:
-      - uam
-
-  kafka:
-    image: confluentinc/cp-kafka:5.2.2
-    ports:
-      - 9092:9092
-      - 9094:9094
-    hostname: kafka
-    environment:
-      KAFKA_ADVERTISED_LISTENERS: LISTENER_DOCKER_INTERNAL://kafka:9094,LISTENER_DOCKER_EXTERNAL://${DOCKER_HOST_IP:-127.0.0.1}:9092
-      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: LISTENER_DOCKER_INTERNAL:PLAINTEXT,LISTENER_DOCKER_EXTERNAL:PLAINTEXT
-      KAFKA_INTER_BROKER_LISTENER_NAME: LISTENER_DOCKER_INTERNAL
-      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
-      KAFKA_BROKER_ID: 1
-      KAFKA_LOG4J_LOGGERS: kafka.controller=INFO,kafka.producer.async.DefaultEventHandler=INFO,state.change.logger=INFO
-      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
-      KAFKA_LOG_RETENTION_HOURS: 168000
-    depends_on:
-      - zookeeper
-    networks:
-      - uam
-
-  minio:
-    image: minio/minio
-    ports:
-      - 9000:9000
-    hostname: minio
-    environment:
-      MINIO_ACCESS_KEY: minio
-      MINIO_SECRET_KEY: minio123
-      MINIO_HTTP_TRACE: /dev/stdout
-    command: server /data
-    networks:
-      - uam
-
-  default-buckets:
-    image: minio/mc
-    entrypoint: >
-      /bin/sh -c "
-      sleep 20;
-      /usr/bin/mc config host add minio http://minio:9000 minio minio123;
-      /usr/bin/mc mb minio/dev-data-raw;
-      /usr/bin/mc policy public minio/dev-data-raw;"
-    depends_on:
-      - minio
-    networks:
-      - uam
-
-  kafka-connect:
-    image: confluentinc/cp-kafka-connect:5.2.2
-    hostname: kafka-connect
-    ports:
-      - 8083:8083
-      - 3030:3030
-    environment:
-      CONNECT_REST_ADVERTISED_HOST_NAME: kafka-connect
-      CONNECT_REST_PORT: 8083
-      CONNECT_BOOTSTRAP_SERVERS: kafka:9094
-      CONNECT_GROUP_ID: primary-etl
-      CONNECT_CONFIG_STORAGE_TOPIC: __primary-etl-configs
-      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1
-      CONNECT_OFFSET_STORAGE_TOPIC: __primary-etl-offsets
-      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1
-      CONNECT_STATUS_STORAGE_TOPIC: __primary-etl-status
-      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1
-      CONNECT_KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
-      CONNECT_KEY_CONVERTER_SCHEMAS_ENABLE: "false"
-      CONNECT_VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
-      CONNECT_VALUE_CONVERTER_SCHEMAS_ENABLE: "false"
-      CONNECT_INTERNAL_KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
-      CONNECT_INTERNAL_KEY_CONVERTER_SCHEMAS_ENABLE: "false"
-      CONNECT_INTERNAL_VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
-      CONNECT_INTERNAL_VALUE_CONVERTER_SCHEMAS_ENABLE: "false"
-      CONNECT_LOG4J_ROOT_LOGLEVEL: INFO
-      CONNECT_LOG4J_LOGGERS: org.apache.kafka.connect.runtime.rest=WARN,org.reflections=ERROR
-      CONNECT_PLUGIN_PATH: /usr/share/java,/etc/kafka-connect/jars
-      AWS_ACCESS_KEY_ID: minio
-      AWS_SECRET_ACCESS_KEY: minio123
-
-    depends_on:
-      - kafka
-      - default-buckets
-    networks:
-      - uam
-
-
-networks:
-  uam:
-    name: uam

tests/test_infrastructure_kinesis_ds.py

@@ -1,69 +0,0 @@
-from unittest import main, TestCase
-import boto3
-import configparser as cp
-import warnings
-import os
-
-
-class TestInfraKinesis(TestCase):
-
-    def setUp(self):
-        self.kinesis_client = boto3.client('kinesis')
-
-        # read configuration
-        # making path universal for running tests from the module / outside
-        cwd = os.getcwd()
-        extra_dot = '.' if cwd.endswith('tests') else ''
-        config_path = extra_dot + "./labs/terraform/terraform.tfvars"
-
-        with open(config_path, 'r') as f:
-            config_string = '[main]\n' + f.read()
-        config = cp.ConfigParser()
-        config.read_string(config_string)
-
-        self.config = {param: (
-            config["main"][param].replace('"', '') if isinstance(config["main"][param], str) else config["main"][param]
-        ) for param in config["main"]}
-
-        warnings.filterwarnings("ignore", category=ResourceWarning, message="unclosed.*<ssl.SSLSocket.*>")
-
-        kinesis_stream_name_pattern = "cryptostock-dev-{account_number}-{student_initials}-{student_index_no}"
-        self.kinesis_stream_name = kinesis_stream_name_pattern.format(account_number=self.config["account_number"],
-                                                                      student_initials=self.config["student_initials"],
-                                                                      student_index_no=self.config["student_index_no"])
-
-    def test_kinesis_data_stream_exists(self):
-        kinesis_streams = self.kinesis_client.list_streams()["StreamNames"]
-
-        find_stream = next((stream for stream in kinesis_streams if stream == self.kinesis_stream_name), None)
-
-        self.assertEqual(find_stream, self.kinesis_stream_name)
-
-    def test_kinesis_data_stream_config(self):
-        expected_no_of_shards = 1
-
-        stream_config = self.kinesis_client.describe_stream(StreamName=self.kinesis_stream_name)
-
-        # check no of shards
-        no_of_shards = len(stream_config["StreamDescription"]["Shards"])
-        self.assertEqual(no_of_shards, expected_no_of_shards)
-
-    def test_kinesis_data_stream_tags(self):
-        tags = self.kinesis_client.list_tags_for_stream(StreamName=self.kinesis_stream_name)["Tags"]
-
-        find_env_tag = next((tag["Value"] for tag in tags if tag.get("Key") == "Environment"), None)
-        self.assertEqual(find_env_tag.upper(), "DEV")
-
-    def test_kinesis_data_stream_monitoring(self):
-        expected_monitors = ['IncomingBytes', 'OutgoingRecords', 'IncomingRecords', 'OutgoingBytes']
-
-        stream_monitoring = \
-            self.kinesis_client.describe_stream(StreamName=self.kinesis_stream_name)["StreamDescription"][
-                "EnhancedMonitoring"]
-
-        current_monitors = next((x["ShardLevelMetrics"] for x in stream_monitoring if x.get("ShardLevelMetrics")), None)
-        self.assertTrue(set(current_monitors).issuperset(set(expected_monitors)))
-
-
-if __name__ == '__main__':
-    main()

tests/test_infrastructure_kinesis_firehose.py

@@ -1,73 +0,0 @@
-from unittest import main, TestCase
-import boto3
-import configparser as cp
-import warnings
-import os
-
-
-class TestInfraKinesisFH(TestCase):
-
-    def setUp(self):
-        self.firehose_client = boto3.client('firehose')
-        self.kinesis_client = boto3.client('kinesis')
-
-        # read configuration
-        # making path universal for running tests from the module / outside
-        cwd = os.getcwd()
-        extra_dot = '.' if cwd.endswith('tests') else ''
-        config_path = extra_dot + "./labs/terraform/terraform.tfvars"
-
-        with open(config_path, 'r') as f:
-            config_string = '[main]\n' + f.read()
-        config = cp.ConfigParser()
-        config.read_string(config_string)
-
-        self.config = {param: (
-            config["main"][param].replace('"', '') if isinstance(config["main"][param], str) else config["main"][param]
-        ) for param in config["main"]}
-
-        warnings.filterwarnings("ignore", category=ResourceWarning, message="unclosed.*<ssl.SSLSocket.*>")
-
-        firehose_stream_name_pattern = "firehose-dev-{account_number}-{student_initials}-{student_index_no}"
-        self.firehose_stream_name = firehose_stream_name_pattern.format(account_number=self.config["account_number"],
-                                                                        student_initials=self.config[
-                                                                            "student_initials"],
-                                                                        student_index_no=self.config[
-                                                                            "student_index_no"])
-
-        kinesis_stream_name_pattern = "cryptostock-dev-{account_number}-{student_initials}-{student_index_no}"
-        self.kinesis_stream_name = kinesis_stream_name_pattern.format(account_number=self.config["account_number"],
-                                                                      student_initials=self.config["student_initials"],
-                                                                      student_index_no=self.config["student_index_no"])
-
-    def test_kinesis_firehose_exists(self):
-        deliver_streams = self.firehose_client.list_delivery_streams()["DeliveryStreamNames"]
-        find_stream = next((stream for stream in deliver_streams if stream == self.firehose_stream_name), None)
-
-        self.assertEqual(find_stream, self.firehose_stream_name)
-
-    def test_kinesis_firehose_source(self):
-        config = self.firehose_client.describe_delivery_stream(DeliveryStreamName=self.firehose_stream_name)
-        source = config["DeliveryStreamDescription"]["Source"]["KinesisStreamSourceDescription"]["KinesisStreamARN"]
-
-        kinesis_stream_config = self.kinesis_client.describe_stream(StreamName=self.kinesis_stream_name)
-        kds_arn = kinesis_stream_config["StreamDescription"]["StreamARN"]
-
-        self.assertEqual(kds_arn, source)
-
-    def test_kinesis_firehose_dest_config(self):
-        config = self.firehose_client.describe_delivery_stream(DeliveryStreamName=self.firehose_stream_name)
-        destination = config["DeliveryStreamDescription"]["Destinations"][0]
-
-        expected_pref = 'raw-zone/stockdata/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/hour=!{timestamp:HH}/'
-        expected_err_pref = 'raw-zone/stockdata_errors/!{firehose:error-output-type}/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/hour=!{timestamp:HH}/'
-
-        dest_prefix = destination.get("S3DestinationDescription", None).get("Prefix", None)
-        dest_errors_prefix = destination.get("S3DestinationDescription", None).get("ErrorOutputPrefix", None)
-
-        self.assertEqual(dest_prefix, expected_pref)
-        self.assertEqual(expected_err_pref, dest_errors_prefix)
-
-
-if __name__ == '__main__':
-    main()

tests/test_infrastructure_s3.py

@@ -1,52 +0,0 @@
-from unittest import main, TestCase
-import boto3
-import configparser as cp
-import warnings
-import os
-
-
-class TestInfraS3(TestCase):
-
-    def setUp(self):
-        self.s3_client = boto3.client('s3')
-
-        # read configuration
-        # making path universal for running tests from the module / outside
-        cwd = os.getcwd()
-        extra_dot = '.' if cwd.endswith('tests') else ''
-        config_path = extra_dot + "./labs/terraform/terraform.tfvars"
-
-        with open(config_path, 'r') as f:
-            config_string = '[main]\n' + f.read()
-        config = cp.ConfigParser()
-        config.read_string(config_string)
-
-        self.config = {param: (
-            config["main"][param].replace('"', '') if isinstance(config["main"][param], str) else config["main"][param]
-        ) for param in config["main"]}
-
-        warnings.filterwarnings("ignore", category=ResourceWarning, message="unclosed.*<ssl.SSLSocket.*>")
-
-        bucket_name_pattern = "datalake-dev-{account_number}-{student_initials}-{student_index_no}"
-        self.bucket_name = bucket_name_pattern.format(account_number=self.config["account_number"],
-                                                      student_initials=self.config["student_initials"],
-                                                      student_index_no=self.config["student_index_no"])
-
-    def test_main_s3_bucket_exists(self):
-        s3_buckets = self.s3_client.list_buckets()["Buckets"]
-
-        find_bucket = next((bucket["Name"] for bucket in s3_buckets if bucket["Name"] == self.bucket_name), None)
-
-        self.assertEqual(find_bucket, self.bucket_name)
-
-    def test_main_s3_bucket_comfig(self):
-        tags = self.s3_client.get_bucket_tagging(
-            Bucket=self.bucket_name
-        )["TagSet"]
-
-        find_env_tag = next((tag["Value"] for tag in tags if tag.get("Key") == "Environment"), None)
-        self.assertEqual(find_env_tag.upper(), "DEV")
-
-
-if __name__ == '__main__':
-    main()