camada_em_nevoa - rafaelglima/smarthealth GitHub Wiki

Camada em Névoa - Apache Kafka (Amazon MSK) e AWS Lambda

Configurando o Amazon MSK (Cluster Apache Kafka)

Embora seja possível utilizar outro serviço similar como o Amazon Data Kinesis Data Stream, optou-se pelo Apache Kafka gerenciado pela AWS uma vez que é possível definir quais as suas subredes, o que torna o ambiente mais próximo de um cenário Fog e multi plataforma.
Foi criado o cluster Amazon MSK utilizando o arquivo via Cloud Formation.

(Arquivo: shdm-item11-msk-sa-east-1.yaml)

Configurações conforme o link: [IoT com MSK] (https://aws.amazon.com/blogs/iot/how-to-integrate-aws-iot-core-with-amazon-msk/)

Criar o tópico no cluster msk antes de enviar para a função lambda

[Autenticação SASL] (https://docs.aws.amazon.com/pt_br/msk/latest/developerguide/msk-password.html)

[Criando topico] (https://docs.aws.amazon.com/pt_br/msk/latest/developerguide/msk-password.html)

sudo apt-get install openjdk-11-jdk
wget https://archive.apache.org/dist/kafka/3.5.1/kafka_2.13-3.5.1.tgz
tar -xzf kafka_2.13-3.5.1.tgz
cd kafka_2.13-3.5.1/libs
wget https://github.com/aws/aws-msk-iam-auth/releases/download/v1.1.1/aws-msk-iam-auth-1.1.1-all.jar
cd ../bin/
touch client_sasl.properties
vi client_sasl.properties

[Adicione as linhas abaixo no arquivo]
security.protocol=SASL_SSL
sasl.mechanism=AWS_MSK_IAM
sasl.jaas.config=software.amazon.msk.auth.iam.IAMLoginModule required;
sasl.client.callback.handler.class=software.amazon.msk.auth.iam.IAMClientCallbackHandler

# Coletar Informações dos Brokers

aws kafka get-bootstrap-brokers --region sa-east-1 --cluster-arn arn:aws:kafka:sa-east-1:904233115303:cluster/shdm-item11-msk-sa-east-1/1bcf05aa-cb7c-4b77-be53-7a59819ad04f-4

# Criar Tópicos Kafka

./kafka-topics.sh --create --bootstrap-server b-1.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096 --command-config client_sasl.properties --replication-factor 2 --partitions 30 --topic topic-oximetria-pulso

./kafka-topics.sh --create --bootstrap-server b-1.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096 --command-config client_sasl.properties --replication-factor 2 --partitions 30 --topic topic-quedas

./kafka-topics.sh --create --bootstrap-server b-1.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096 --command-config client_sasl.properties --replication-factor 2 --partitions 30 --topic topic-pressao-arterial

./kafka-topics.sh --create --bootstrap-server b-1.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096 --command-config client_sasl.properties --replication-factor 2 --partitions 30 --topic topic-glicemia

./kafka-topics.sh --create --bootstrap-server b-1.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096 --command-config client_sasl.properties --replication-factor 2 --partitions 30 --topic topic-temperatura-ambiente

./kafka-topics.sh --create --bootstrap-server b-1.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096 --command-config client_sasl.properties --replication-factor 2 --partitions 30 --topic topic-frequencia-cardiaca

./kafka-topics.sh --create --bootstrap-server b-1.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096 --command-config client_sasl.properties --replication-factor 2 --partitions 30 --topic topic-frequencia-respiratoria

./kafka-topics.sh --create --bootstrap-server b-1.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096 --command-config client_sasl.properties --replication-factor 2 --partitions 30 --topic topic-umidade-relativa

./kafka-topics.sh --create --bootstrap-server b-1.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096 --command-config client_sasl.properties --replication-factor 2 --partitions 30 --topic topic-temperatura-corporal

./kafka-topics.sh --create --bootstrap-server b-1.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096 --command-config client_sasl.properties --replication-factor 2 --partitions 30 --topic topic-geolocalizacao

# Apagar Tópicos

./kafka-topics.sh --bootstrap-server b-2.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096,b-1.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096 --command-config client_sasl.properties --delete --topic topic-oximetria-pulso

./kafka-topics.sh --bootstrap-server b-2.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096,b-1.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096 --command-config client_sasl.properties --delete --topic topic-quedas

./kafka-topics.sh --bootstrap-server b-2.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096,b-1.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096 --command-config client_sasl.properties --delete --topic topic-pressao-arterial

./kafka-topics.sh --bootstrap-server b-2.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096,b-1.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096 --command-config client_sasl.properties --delete --topic topic-glicemia

./kafka-topics.sh --bootstrap-server b-2.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096,b-1.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096 --command-config client_sasl.properties --delete --topic topic-temperatura-ambiente

./kafka-topics.sh --bootstrap-server b-2.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096,b-1.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096 --command-config client_sasl.properties --delete --topic topic-frequencia-cardiaca

./kafka-topics.sh --bootstrap-server b-2.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096,b-1.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096 --command-config client_sasl.properties --delete --topic topic-frequencia-respiratoria

./kafka-topics.sh --bootstrap-server b-2.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096,b-1.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096 --command-config client_sasl.properties --delete --topic topic-umidade-relativa

./kafka-topics.sh --bootstrap-server b-2.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096,b-1.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096 --command-config client_sasl.properties --delete --topic topic-temperatura-corporal

./kafka-topics.sh --bootstrap-server b-2.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096,b-1.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096 --command-config client_sasl.properties --delete --topic topic-geolocalizacao

# Descrever e Listar Tópicos

./kafka-topics.sh --bootstrap-server b-2.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096,b-1.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096 --command-config client_sasl.properties --describe --topic topic-frequencia-cardiaca

./kafka-topics.sh --bootstrap-server b-2.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096,b-1.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096 --command-config client_sasl.properties --list

# Verificar Quantidade de Mensagens no Tópico Kafka

export KAFKA_OPTS=-Djava.security.auth.login.config=/home/ubuntu/msk/kafka_2.13-3.5.1/bin/users_jaas.conf

./kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list b-2.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096,b-1.shdmitem11msksaeast1.na7egv.c4.kafka.sa-east-1.amazonaws.com:9096 --topic topic-frequencia-cardiaca --time -1 --command-config client_sasl.properties

Observação: Importante definir a quantidade de partições e o caminho do da variável de ambiente.

Criar a rule para rotear mensagens do IoT Core pro MSK.

${get_secret('AmazonMSK_chave', 'SecretString', 'username', 'arn:aws:iam::730335447253:role/iot-role-msk')}

${get_secret('AmazonMSK_chave', 'SecretString', 'password', 'arn:aws:iam::730335447253:role/iot-role-msk')}

Criar o Bucket S3 para representar o repositório (Data Lake) de imagens / arquivos de sistemas médicos que vieram a partir da camada de Usuários

(Arquivo: shdm-item14-buckets3-fog.yaml)

Antes de criar o bucket, crie uma chave no KMS e altere no arquivo acima (KMS-KEY-ARN). Ao criar essa chave deve-se atribuir a permissão para a role da função lambda (via console do KMS).
Após a criação do Bucket será necessário criar uma função Lambda que será acionada mediante inserção de arquivos no Bucket que foram inseridos pelo Amazon Kinesis Firehose.
Essa função Lambda será responsável pela transformação dos dados e envio para a Camada em Nuvem.

Criando a função lambda referente ao envio de arquivos para o Cluster IPFS e informações para o Cluster HapiFhir e Blockchain

mkdir lambda-fog-ipfs
cd lambda-fog-ipfs
mkdir venv
cd venv
virtualenv lambda-fog-ipfs
source lambda-fog-ipfs/bin/activate
cd ..
pip install pytest
pip install chalice
pip install httpx
pip install httpcore
pip install urllib3==1.26.18
pip install requests==2.28.2
pip install boto3
pip install botocore
pip install ipfshttpclient==0.4.12
pip install ipfscluster==0.2.0
chalice new-project
[definir nome lambda-fog-ipfs e escolher S3 Event Handler]
[confira a codificação]
cd lambda-fog-ipfs
chalice deploy

Criando a Layer para adicionar à Função Lambda

mkdir python
pip cache purge
python -m pip install httpx -t python/ && \
python -m pip install httpcore -t python/ && \
python -m pip install urllib3==1.26.18 -t python/ && \
python -m pip install requests==2.28.2 -t python/ && \
python -m pip install boto3 -t python/ && \
python -m pip install botocore -t python/ && \
python -m pip install ipfshttpclient==0.4.12 -t python/ && \
python -m pip install ipfscluster==0.2.0 -t python/

zip -r layer-fog-ipfs.zip python

aws lambda publish-layer-version --layer-name layer-fog-ipfs --zip-file fileb://layer-fog-ipfs.zip --compatible-runtimes python3.10 --region sa-east-1

Definindo as subredes e layer da implantação da função Lambda no arquivo .chalice/config.json

Deve-se alterar os ids das subredes, o id do security group e a layer com sua versão.

{
  "version": "2.0",
  "app_name": "lambda-fog-ipfs",
  "stages": {
    "dev": {
      "api_gateway_stage": "api",
      "subnet_ids": ["subnet-07bc3f3031ccf1600","subnet-0651f5557b6a0a0bc","subnet-095efa24bfa61fdef"],
      "layers": ["arn:aws:lambda:sa-east-1:730335447253:layer:layer-fog-ipfs:1"]
    }
  }
}

Criar o processo para receber dados dos sensores IoT da Camada Usuários

Após a criação do Amazon MSK (Apache Kafka) deve-se criar as Roles para cada tipo de sensor do IoT Device Simulator no IoT Core (e criar a VPC Destination) para o envio dos dados dos sensores para o Amazon MSK.
Além disso, deve-se criar no Amazon MSK a integração de Delivery para AWS Lambda via Pipes do EventBridge.

----

Criando a função lambda referente ao envio de medições para o Cluster HapiFhir e Blockchain

mkdir lambda-fog-ipfs
cd lambda-fog-iot
mkdir venv
cd venv
virtualenv lambda-fog-iot
source lambda-fog-iot/bin/activate
cd ..
pip install pytest
pip install chalice
pip install httpx
pip install httpcore
pip install urllib3==1.26.18
pip install requests==2.28.2
pip install boto3
pip install botocore
chalice new-project
[definir nome lambda-fog-iot e escolher Lambda Functions only]
[confira a codificação]
cd lambda-fog-iot
chalice deploy

Criando a Layer para adicionar à Função Lambda

mkdir python
pip cache purge
python -m pip install httpx -t python/ && \
python -m pip install httpcore -t python/ && \
python -m pip install urllib3==1.26.18 -t python/ && \
python -m pip install requests==2.28.2 -t python/ && \
python -m pip install boto3 -t python/ && \
python -m pip install botocore -t python/

zip -r layer-fog-iot.zip python

aws lambda publish-layer-version --layer-name layer-fog-iot --zip-file fileb://layer-fog-iot.zip --compatible-runtimes python3.10 --region sa-east-1

Definindo as subredes e layer da implantação da função Lambda no arquivo .chalice/config.json

Deve-se alterar os ids das subredes, o id do security group e a layer com sua versão.

{
  "version": "2.0",
  "app_name": "lambda-fog-iot",
  "stages": {
    "dev": {
      "api_gateway_stage": "api",
      "subnet_ids": ["subnet-07bc3f3031ccf1600","subnet-0651f5557b6a0a0bc","subnet-095efa24bfa61fdef"],
      "layers": ["arn:aws:lambda:sa-east-1:730335447253:layer:layer-fog-iot:1"]
    }
  }
}

Referências

[Criar Cluster Amazon MSK] (https://docs.aws.amazon.com/pt_br/msk/latest/developerguide/create-cluster.html)

[Criar um tópico no Amazon MSK] (https://docs.aws.amazon.com/pt_br/msk/latest/developerguide/create-topic.html)

[IPFS Cluster API] (https://clearos.gitlab.io/clearfoundation/py-ipfs-cluster-api/client.html)

[Criando uma Layer] (https://repost.aws/knowledge-center/lambda-import-module-error-python)

[Arquivo config.json do Chalice] (https://chalice-fei.readthedocs.io/en/latest/topics/configfile.html)

[Boto3] (https://boto3.amazonaws.com/v1/documentation/api/latest/guide/s3.html)

Criando Policy e Role Iot-MSK para VPC Destination