Loads of updates to extensions and their documentation
This commit is contained in:
frikky
@@ -1,2 +0,0 @@
|
||||
# AWS Lambda forwarder to Shuffle
|
||||
This function is made to forward S3 notifications to Shuffle to run a workflow when an object is made or updated.
|
||||
@@ -1,16 +0,0 @@
|
||||
#GOOS=linux go build main.go
|
||||
zip s3_function.zip s3_function.py
|
||||
|
||||
WEBHOOK=https://shuffler.io/api/v1/hooks/webhook_eccf47b1-8f6a-49fc-b2b8-383365a22353
|
||||
ROLE=arn:aws:iam::202262580068:role/service-role/shuffle-forwarder
|
||||
REGION=us-east-1
|
||||
BUCKETNAME=helo
|
||||
|
||||
aws lambda create-function \
|
||||
--role $ROLE \
|
||||
--region $REGION \
|
||||
--function-name shuffler-webhook-forwarder-3 \
|
||||
--zip-file fileb://s3_function.zip \
|
||||
--runtime python3.9 \
|
||||
--environment Variables={SHUFFLE_WEBHOOK=$WEBHOOK} \
|
||||
--handler lambda_handler
|
||||
@@ -1,26 +0,0 @@
|
||||
import json
|
||||
import urllib.parse
|
||||
import requests
|
||||
import os
|
||||
|
||||
print('Loading function')
|
||||
|
||||
def lambda_handler(event, context):
|
||||
#print("Received event: " + json.dumps(event, indent=2))
|
||||
|
||||
# Get the object from the event and show its content type
|
||||
bucket = event['Records'][0]['s3']['bucket']['name']
|
||||
|
||||
print(type(event))
|
||||
print("Getting bucket: %s" % bucket)
|
||||
webhook = os.environ.get("SHUFFLE_WEBHOOK")
|
||||
if not webhook:
|
||||
return "No webhook environment defined: SHUFFLE_WEBHOOK"
|
||||
|
||||
ret = requests.post(webhook, json=event["Records"][0])
|
||||
if ret.status_code != 200:
|
||||
return "Bad status code for webhook: %d" % ret.status_code
|
||||
|
||||
print("Status code: %d\nData: %s" % (ret.status_code, ret.text))
|
||||
|
||||
# response = s3.get_object(Bucket=bucket, Key=key)
|
||||
@@ -1,4 +0,0 @@
|
||||
```
|
||||
curl -sSL https://raw.githubusercontent.com/bitnami/bitnami-docker-kafka/master/docker-compose.yml > docker-compose.yml
|
||||
docker-compose up -d
|
||||
```
|
||||
@@ -1,16 +0,0 @@
|
||||
version: '2'
|
||||
services:
|
||||
zookeeper:
|
||||
image: wurstmeister/zookeeper
|
||||
ports:
|
||||
- "2181:2181"
|
||||
kafka:
|
||||
build: .
|
||||
ports:
|
||||
- "9092"
|
||||
environment:
|
||||
DOCKER_API_VERSION: 1.40
|
||||
KAFKA_ADVERTISED_HOST_NAME: 192.168.193.140
|
||||
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
|
||||
volumes:
|
||||
- /var/run/docker.sock:/var/run/docker.sock
|
||||
@@ -1,87 +0,0 @@
|
||||
import json
|
||||
import os
|
||||
import requests
|
||||
from time import sleep
|
||||
from kafka import KafkaProducer, KafkaConsumer
|
||||
import kafka
|
||||
|
||||
shuffle_url = os.getenv("SHUFFLE_URL")
|
||||
shuffle_apikey = os.getenv("SHUFFLE_APIKEY")
|
||||
shuffle_workflow = os.getenv("SHUFFLE_WORKFLOW")
|
||||
|
||||
headers = {"Authorization": "Bearer %s" % shuffle_apikey}
|
||||
topic = "workflow_%s" % shuffle_workflow
|
||||
group = "testing"
|
||||
server = "localhost:9092"
|
||||
def produce():
|
||||
try:
|
||||
#group_id=group,
|
||||
#value_serializer=lambda x: json.dumps(x).encode('utf-8')
|
||||
producer = KafkaProducer(
|
||||
bootstrap_servers=[server],
|
||||
)
|
||||
except kafka.errors.NoBrokersAvailable as e:
|
||||
print(f"Error with producer: {e}")
|
||||
sleep(5)
|
||||
produce()
|
||||
return
|
||||
except ValueError as e:
|
||||
print(f"ValuError with producer: {e}")
|
||||
sleep(5)
|
||||
produce()
|
||||
return
|
||||
|
||||
|
||||
for i in range(15):
|
||||
producer.send(topic, json.dumps({"some": i, "data": "luuuuul"}))
|
||||
|
||||
producer.flush()
|
||||
producer.close()
|
||||
exit()
|
||||
|
||||
#print(f"Adding data {data}")
|
||||
|
||||
#try:
|
||||
# ret = producer.send(topic, value=data)
|
||||
# print(ret.get())
|
||||
# producer.flush()
|
||||
|
||||
# #future = producer.send('foobar', b'another_message')
|
||||
# #result = future.get(timeout=60)
|
||||
# #print(result)
|
||||
|
||||
#except kafka.errors.KafkaTimeoutError as e:
|
||||
# print("Kafka error: %s" % e)
|
||||
# continue
|
||||
|
||||
def consume():
|
||||
print("Starting consumer")
|
||||
#group_id=group,
|
||||
#auto_offset_reset="earliest",
|
||||
#enable_auto_commit=True,
|
||||
consumer = KafkaConsumer(
|
||||
topic,
|
||||
bootstrap_servers=[server],
|
||||
auto_offset_reset='earliest',
|
||||
max_poll_records=2,
|
||||
)
|
||||
#value_deserializer=lambda x: json.loads(x.decode('utf-8'))
|
||||
|
||||
consumer.poll()
|
||||
#consumer.seek_to_beginning()
|
||||
|
||||
print(f"Getting data from topic {topic}")
|
||||
for message in consumer:
|
||||
#message = message.value
|
||||
#message = message
|
||||
print("MSG: ", message)
|
||||
ret = requests.post("%s/api/v1/%s/execute" % shuffle_url, headers=headers, data=message)
|
||||
print(ret.status_code)
|
||||
print(ret.text)
|
||||
|
||||
if __name__ == "__main__":
|
||||
if topic == "workflow_None":
|
||||
topic = "testing"
|
||||
print("Starting producer on %s for topic %s" % (server, topic))
|
||||
produce()
|
||||
#consume()
|
||||
@@ -1,2 +0,0 @@
|
||||
kafka
|
||||
kafka-python
|
||||
@@ -0,0 +1,23 @@
|
||||
FROM python:3.9.4-alpine as base
|
||||
|
||||
FROM base as builder
|
||||
|
||||
RUN mkdir /install
|
||||
WORKDIR /install
|
||||
|
||||
FROM base
|
||||
RUN apk add g++
|
||||
|
||||
COPY --from=builder /install /usr/local
|
||||
COPY requirements.txt /requirements.txt
|
||||
RUN pip3 install -r /requirements.txt
|
||||
|
||||
|
||||
RUN mkdir /app
|
||||
WORKDIR /app
|
||||
COPY requirements.txt /app/requirements.txt
|
||||
RUN python3 -m pip install -r /app/requirements.txt
|
||||
|
||||
COPY sub.py /app/sub.py
|
||||
|
||||
CMD ["python3", "sub.py"]
|
||||
@@ -0,0 +1,9 @@
|
||||
version: '3'
|
||||
services:
|
||||
zmq:
|
||||
image: ghcr.io/frikky/shuffle-zmq:latest
|
||||
environment:
|
||||
- ZMQ_HOSTNAME=localhost
|
||||
- ZMQ_PORT=50000
|
||||
- ZMQ_FORWARD_URL=https://shuffler.io/api/v1/hooks/webhook_e09bea36-9976-1421-82bc-b8764ca83c1e
|
||||
restart: unless-stopped
|
||||
@@ -0,0 +1,2 @@
|
||||
pyzmq
|
||||
requests
|
||||
@@ -0,0 +1,57 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
print("Running imports")
|
||||
import sys
|
||||
import zmq
|
||||
import json
|
||||
import time
|
||||
import pprint
|
||||
import os
|
||||
import sys
|
||||
import requests
|
||||
|
||||
forward_url = os.getenv("ZMQ_FORWARD_URL", "")
|
||||
print("Checking forward url (ZMQ_FORWARD_URL): %s" % forward_url)
|
||||
def handle_hook(data):
|
||||
ret = requests.post(forward_url, json=data)
|
||||
print(ret.text)
|
||||
print(ret.status_code)
|
||||
|
||||
def main():
|
||||
host = os.getenv("ZMQ_HOST", "localhost")
|
||||
port = os.getenv("ZMQ_PORT", "50000")
|
||||
|
||||
if len(forward_url) == 0:
|
||||
print("Failed to start - define ZMQ_FORWARD_URL for webhook forwarder")
|
||||
exit(0)
|
||||
|
||||
print("Starting connection setup to %s:%s" % (host, port))
|
||||
context = zmq.Context()
|
||||
socket = context.socket(zmq.SUB)
|
||||
socket.connect ("tcp://%s:%s" % (host, port))
|
||||
socket.setsockopt(zmq.SUBSCRIBE, b'')
|
||||
|
||||
poller = zmq.Poller()
|
||||
poller.register(socket, zmq.POLLIN)
|
||||
|
||||
print("Starting zmq check for %s:%s" % (host, port))
|
||||
while True:
|
||||
socks = dict(poller.poll(timeout=None))
|
||||
if socket in socks and socks[socket] == zmq.POLLIN:
|
||||
message = socket.recv()
|
||||
#print(message)
|
||||
topic, s, m = message.decode('utf-8').partition(" ")
|
||||
|
||||
d = json.loads(m)
|
||||
try:
|
||||
# print test if you want status (heartbeat)
|
||||
test = d["status"]
|
||||
except KeyError:
|
||||
handle_hook(d)
|
||||
|
||||
time.sleep(1)
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("In init ")
|
||||
main()
|
||||
Reference in New Issue
Block a user