marker/shuffle-cracked
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Started fixing automated authentication from apps

Browse Source
This commit is contained in:
frikky
2020-05-12 18:25:20 +02:00
parent 4748c0fab6
commit d27650d76b
12 changed files with 84 additions and 2230 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+19 -34
View File
@@ -16,6 +16,11 @@ Documentation can be found on https://shuffler.io/docs/about or in your own inst
* App creator for [OpenAPI](https://github.com/frikky/OpenAPI-security-definitions) * App creator for [OpenAPI](https://github.com/frikky/OpenAPI-security-definitions)
* Easy to learn Python library for custom apps * Easy to learn Python library for custom apps
## In the works
* Full OpenAPI support with authentication schemes in App creator
* User run statistics - Dashboard
* Debug view for manual executions
### Setup - Local ### Setup - Local
Frontend - requires [npm](https://nodejs.org/en/download/)/[yarn](https://yarnpkg.com/lang/en/docs/install/#debian-stable)/your preferred manager. Runs independently from backend - edit frontend/src/App.yaml to change from localhost to prod setting. Frontend - requires [npm](https://nodejs.org/en/download/)/[yarn](https://yarnpkg.com/lang/en/docs/install/#debian-stable)/your preferred manager. Runs independently from backend - edit frontend/src/App.yaml to change from localhost to prod setting.
```bash ```bash
@@ -36,21 +41,18 @@ go run *.go
Below is the folder structure with a short explanation Below is the folder structure with a short explanation
```bash ```bash
├── README.md # :) ├── README.md # :)
├── deploy.sh # Simple oneliner script to build and deploy the code to gcloud
├── backend # Contains directly backend related code. Go with sh tests ├── backend # Contains directly backend related code. Go with sh tests
├── frontend # Contains frontend code. ReactJS and cytoscape. Horrible code :) ├── frontend # Contains frontend code. ReactJS and cytoscape. Horrible code :)
├── app_gen # Contains code generation projects for OpenAPI or PythonLib -> Shuffler app ├── app_gen # Contains code generation projects for OpenAPI or PythonLib -> Shuffler app
├── functions # Contains google cloud function code mainly. ├── functions # Contains google cloud function code mainly.
│   ├── apps # Some of the existing apps, manually made mostly
│   ├── generated_apps # Some of the autogenerated apps
│   ├── newworker # The worker that handles a workflow as a google cloud function
│   ├── static_baseline.py # Static code used by stitcher.go to generate code │   ├── static_baseline.py # Static code used by stitcher.go to generate code
│   ├── stitcher.go # Attempts to stitch together an app and deploy it to cloud functions and (TBD: Docker hub) │   ├── stitcher.go # Attempts to stitch together an app - part of backend now
│   ├── onprem # Code for onprem solutions
│  │   ├── Orborus # Distributes execution locations
│  │   ├── Worker # Runs a workflow
│ │   └── App_SDK # Backend of individual apps
│   └── triggers # Custom triggers used in https://shuffler.io/workflows │   └── triggers # Custom triggers used in https://shuffler.io/workflows
│   ├── onprem # All code for onprem solutions (https://shuffler.io/docs/hybrid for short doc) _mostly_ reflects google cloud. Should be deprecated somehow and use the same code. └ docker-compose.yml # Used for deployments
├── openintegrationhub # Here to remind me that openintegrationhub is a thing
├── legacy # Legacy README. Contains A LOT of useful information about what I found with WALKOFF
└── tmp # Some legacy code, not yet ready to be removed
``` ```
# Architecture # Architecture
@@ -61,34 +63,18 @@ A basic image of how everything fits together, including legacy (left side)
GCP was chosen because why not use the best thingies. "Serverless" \o/ GCP was chosen because why not use the best thingies. "Serverless" \o/
```bash ```bash
├── languages ├── languages
│   ├── Go # I like go, which is why go. │   ├── Go # I like go, which is why go.
│   ├── Python3.7 # 3.7 specifically because of f-strings and 2.7 deprecation in 2020 │   ├── Python3.7 # 3.7 specifically because of f-strings and 2.7 deprecation in 2020
│   ├── Javascript # Frontend stuff. Uses ReactJS + Cytoscape for visualization │   ├── Javascript # Frontend stuff. Uses ReactJS + Cytoscape for visualization
│   ├── sh/Bash # Basic testing and some deployment stuff │   ├── sh/Bash # Basic testing and some deployments
├── gcloud ├── gcloud
│   ├── appengine # Hosting frontend and backend. Currently on a free plan which is nice :) │   ├── datastore # TODO: Move away from this
│   ├── cloud functions # Runs the "apps", "triggers" and other things
│   ├── datastore # Database - TODO before live: Move to firebase
│   ├── storage # Save datablobs and information before deployment
│   ├── pubsub # Used to instantly run cloud functions
│   ├── scheduler # Schedules can be triggers
├── onprem ├── onprem
│   ├── Docker # Runs the same cloud functions. I didn't like the thought of proxies │   ├── Docker # Runs the same cloud functions. I didn't like the thought of proxies
``` ```
# Current focus(es) AKA todo
1. Make workflows work 99%+ of the time. This is a challenge with onprem + cloud stuff. Cloud sometimes breaks currently because of workers
2. Add user run statistics (e.g. how many runs of each workflow, how many failures etc.)
3. X - Fix OpenAPI app generator
4. Fix error overview in workflows
5. X - Better GUI (improved, but not good)
6. Have default workflows
# How to Add a trigger / custom thing
1. Add it to TriggersView in AngularWorkflow.js
2. Add it to RightSideBar for trigger
# Migration # Migration
Shuffle was initially built for cloud and SaaS, and a lot broke when it was moved to local execution.
There will be a major overhaul to the backend specifically. I'm currently moving and updating the following: There will be a major overhaul to the backend specifically. I'm currently moving and updating the following:
- Create dockerfiles and a single runscript - Create dockerfiles and a single runscript
- * App creator - (Cloud function -> Docker) - * App creator - (Cloud function -> Docker)
@@ -97,14 +83,13 @@ There will be a major overhaul to the backend specifically. I'm currently moving
- * Dockerfiles - Load the ones that are in workflows with a new version - * Dockerfiles - Load the ones that are in workflows with a new version
- * Docker-compose- Frontend, backend, db & orborus - * Docker-compose- Frontend, backend, db & orborus
- * Configuration - Write setup documentation - Did for docker - * Configuration - Write setup documentation - Did for docker
- * Remove orborus? Can deploy straight, but that would be weird - Won't do this yet
- Workflows - IMPORT DEFAULT WORKFLOWS - Create some towards e.g. TheHive & MISP. - Workflows - IMPORT DEFAULT WORKFLOWS - Create some towards e.g. TheHive & MISP.
- Documentation - General documentation /docs rewrite - Documentation - General documentation /docs rewrite
- API doc - 1. In Shuffle. 2. In e.g. python - API doc - 1. In Shuffle. 2. In e.g. python
- Remove orborus? Can deploy straight, but that would be weird.
- Add secret to orborus - Add secret to orborus
- Change workflow name - Change workflow name
- Remove registration and add user screen - Fix scheduler
- Add external and internal hostname for orborus & worker
``` ```
# 1. export DATASTORE_EMULATOR_HOST=0.0.0.0:8000 # 1. export DATASTORE_EMULATOR_HOST=0.0.0.0:8000
frontend/src/AngularWorkflow.js
+65 -26
View File
File diff suppressed because one or more lines are too long
functions/newworker/Dockerfile
-25
View File
@@ -1,25 +0,0 @@
FROM python:3.7-alpine as base
FROM base as builder
RUN apk --no-cache add --update alpine-sdk
RUN mkdir /install
WORKDIR /install
COPY ./worker/requirements.txt /requirements.txt
RUN git clone "https://github.com/aio-libs/aioredis.git"
RUN pip install --prefix="/install" ./aioredis
RUN pip install --prefix="/install" --no-deps asteval
RUN pip install --prefix="/install" six
RUN pip install --prefix="/install" -r /requirements.txt
FROM base
COPY --from=builder /install /usr/local
COPY ./umpire/common /app/common
COPY ./worker /app/worker
WORKDIR /app
CMD python -m worker.worker
functions/newworker/async_logger.py
-253
View File
@@ -1,253 +0,0 @@
import logging
import asyncio
import sys
import warnings
from logging import StreamHandler, DEBUG, INFO, ERROR, WARNING, CRITICAL, raiseExceptions
class AsyncHandler(StreamHandler):
""" An async wrapper around logging.StreamHandler for async log streams like Redis PUB/SUB"""
def __init__(self, stream=None, loop=None):
"""
Initialize the handler.
If stream is not specified, sys.stderr is used.
"""
super().__init__(stream)
self.loop = loop
async def flush(self):
"""
Flushes the stream.
"""
await self.stream.flush()
async def emit(self, record):
"""
Emit a record.
If a formatter is specified, it is used to format the record.
The record is then written to the stream with a trailing newline. If
exception information is present, it is formatted using
traceback.print_exception and appended to the stream. If the stream
has an 'encoding' attribute, it is used to determine how to do the
output to the stream.
"""
try:
msg = self.format(record)
await self.stream.write(msg + self.terminator)
await self.flush()
except Exception:
self.handleError(record)
async def handle(self, record):
"""
Conditionally emit the specified logging record.
Emission depends on filters which may have been added to the handler.
Wrap the actual emission of the record with acquisition/release of
the I/O thread lock. Returns whether the filter passed the record for
emission.
"""
rv = self.filter(record)
if rv:
self.acquire()
try:
await self.emit(record)
finally:
self.release()
return rv
async def close(self):
if self.stream is not None:
await self.flush()
await self.stream.close()
super().close()
class AsyncLogger(logging.Logger):
""" An async wrapper around logging.Logger for async log streams like Redis PUB/SUB """
def __init__(self, name, level=logging.ERROR, loop=asyncio.get_event_loop()):
super().__init__(name, level=level)
self.loop = loop
async def debug(self, msg, *args, **kwargs):
"""
Log 'msg % args' with severity 'DEBUG'.
To pass exception information, use the keyword argument exc_info with
a true value, e.g.
logger.debug("Houston, we have a %s", "thorny problem", exc_info=1)
"""
if self.isEnabledFor(DEBUG):
await self._log(DEBUG, msg, args, **kwargs)
async def info(self, msg, *args, **kwargs):
"""
Log 'msg % args' with severity 'INFO'.
To pass exception information, use the keyword argument exc_info with
a true value, e.g.
logger.info("Houston, we have a %s", "interesting problem", exc_info=1)
"""
if self.isEnabledFor(INFO):
await self._log(INFO, msg, args, **kwargs)
async def warning(self, msg, *args, **kwargs):
"""
Log 'msg % args' with severity 'WARNING'.
To pass exception information, use the keyword argument exc_info with
a true value, e.g.
logger.warning("Houston, we have a %s", "bit of a problem", exc_info=1)
"""
if self.isEnabledFor(WARNING):
await self._log(WARNING, msg, args, **kwargs)
async def warn(self, msg, *args, **kwargs):
warnings.warn("The 'warn' method is deprecated, use 'warning' instead", DeprecationWarning, 2)
await self.warning(msg, *args, **kwargs)
async def error(self, msg, *args, **kwargs):
"""
Log 'msg % args' with severity 'ERROR'.
To pass exception information, use the keyword argument exc_info with
a true value, e.g.
logger.error("Houston, we have a %s", "major problem", exc_info=1)
"""
if self.isEnabledFor(ERROR):
await self._log(ERROR, msg, args, **kwargs)
async def exception(self, msg, *args, exc_info=True, **kwargs):
"""
Convenience method for logging an ERROR with exception information.
"""
await self.error(msg, *args, exc_info=exc_info, **kwargs)
async def critical(self, msg, *args, **kwargs):
"""
Log 'msg % args' with severity 'CRITICAL'.
To pass exception information, use the keyword argument exc_info with
a true value, e.g.
logger.critical("Houston, we have a %s", "major disaster", exc_info=1)
"""
if self.isEnabledFor(CRITICAL):
await self._log(CRITICAL, msg, args, **kwargs)
fatal = critical
async def log(self, level, msg, *args, **kwargs):
"""
Log 'msg % args' with the integer severity 'level'.
To pass exception information, use the keyword argument exc_info with
a true value, e.g.
logger.log(level, "We have a %s", "mysterious problem", exc_info=1)
"""
if not isinstance(level, int):
if raiseExceptions:
raise TypeError("level must be an integer")
else:
return
if self.isEnabledFor(level):
await self._log(level, msg, args, **kwargs)
async def _log(self, level, msg, args, exc_info=None, extra=None, stack_info=False):
"""
Low-level logging routine which creates a LogRecord and then calls
all the handlers of this logger to handle the record.
"""
sinfo = None
if logging._srcfile:
try:
fn, lno, func, sinfo = self.findCaller(stack_info)
except ValueError:
fn, lno, func = "(unknown file)", 0, "(unknown function)"
else:
fn, lno, func = "(unknown file)", 0, "(unknown function)"
if exc_info:
if isinstance(exc_info, BaseException):
exc_info = (type(exc_info), exc_info, exc_info.__traceback__)
elif not isinstance(exc_info, tuple):
exc_info = sys.exc_info()
record = self.makeRecord(self.name, level, fn, lno, msg, args,
exc_info, func, extra, sinfo)
await self.handle(record)
async def handle(self, record):
"""
Call the handlers for the specified record.
This method is used for unpickled records received from a socket, as
well as those created locally. Logger-level filtering is applied.
"""
if (not self.disabled) and self.filter(record):
await self.callHandlers(record)
async def callHandlers(self, record):
"""
Pass a record to all relevant handlers.
Loop through all handlers for this logger and its parents in the
logger hierarchy. If no handler was found, output a one-off error
message to sys.stderr. Stop searching up the hierarchy whenever a
logger with the "propagate" attribute set to zero is found - that
will be the last logger whose handlers are called.
"""
c = self
found = 0
while c:
for hdlr in c.handlers:
found += 1
if record.levelno >= hdlr.level:
await hdlr.handle(record)
if not c.propagate:
c = None
else:
c = c.parent
if found == 0:
if logging.lastResort:
if record.levelno >= logging.lastResort.level:
logging.lastResort.handle(record)
elif logging.raiseExceptions and not self.manager.emittedNoHandlerWarning:
sys.stderr.write("No handlers could be found for logger"
" \"%s\"\n" % self.name)
self.manager.emittedNoHandlerWarning = True
async def shutdown(self):
"""
Perform any cleanup actions in the logging system (e.g. flushing
buffers).
Should be called at application exit.
"""
for handler in reversed(self.handlers):
# errors might occur, for example, if files are locked
# we just ignore them if raiseExceptions is not set
try:
if handler:
try:
# await handler.acquire() # do we need to lock?
await handler.flush()
await handler.close()
except (OSError, ValueError):
# Ignore errors which might be caused
# because handlers have been closed but
# references to them are still around at
# application exit.
pass
except Exception:
pass
finally:
# handler.release() # We need to release if we decide to lock I guess
self.removeHandler(handler)
except Exception: # ignore everything, as we're shutting down
pass
functions/newworker/config.py
-73
View File
@@ -1,73 +0,0 @@
import logging
from pathlib import Path
import os
logging.basicConfig(level=logging.INFO, format="{asctime} - {name} - {levelname}:{message}", style='{')
logger = logging.getLogger("WALKOFF")
CONFIG_PATH = (Path(__file__).parent / "config.ini").resolve()
def sint(value, default):
if not isinstance(default, int):
raise TypeError("Default value must be of integer type")
try:
return int(value)
except (TypeError, ValueError):
return default
def sfloat(value, default):
if not isinstance(default, int):
raise TypeError("Default value must be of float type")
try:
return float(value)
except (TypeError, ValueError):
return default
class Config:
# Worker options
WORKER_TIMEOUT = os.environ.get("WORKER_TIMEOUT", "30")
API_GATEWAY_URI = os.environ.get("API_GATEWAY_URI", "http://localhost:8001")
WALKOFF_USERNAME = os.environ.get("WALKOFF_USERNAME", '')
WALKOFF_PASSWORD = os.environ.get("WALKOFF_PASSWORD", '')
# Umpire options
APPS_PATH = os.getenv("APPS_PATH", "../apps")
APP_REFRESH = os.getenv("APP_REFRESH", "60")
SWARM_NETWORK = os.getenv("SWARM_NETWORK", "walkoff_default")
APP_PREFIX = os.getenv("APP_PREFIX", "walkoff_app")
STACK_PREFIX = os.getenv("STACK_PREFIX", "walkoff")
DOCKER_REGISTRY = os.getenv("DOCKER_REGISTRY", "127.0.0.1:5000")
UMPIRE_HEARTBEAT = os.getenv("UMPIRE_HEARTBEAT", "1")
# Redis options
REDIS_URI = os.getenv("REDIS_URI", "redis://192.168.239.145:6379")
REDIS_EXECUTING_WORKFLOWS = os.getenv("REDIS_EXECUTING_WORKFLOWS", "executing-workflows")
REDIS_PENDING_WORKFLOWS = os.getenv("REDIS_PENDING_WORKFLOWS", "pending-workflows")
REDIS_ABORTING_WORKFLOWS = os.getenv("REDIS_ABORTING_WORKFLOWS", "aborting-workflows")
REDIS_ACTIONS_IN_PROCESS = os.getenv("REDIS_ACTIONS_IN_PROCESS", "actions-in-process")
REDIS_WORKFLOW_QUEUE = os.getenv("REDIS_WORKFLOW_Q", "workflow-queue")
REDIS_WORKFLOWS_IN_PROCESS = os.getenv("REDIS_WORKFLOWS_IN_PROCESS", "workflows-in-process")
REDIS_WORKFLOW_GROUP = os.getenv("REDIS_WORKFLOW_GROUP", "workflow-group")
REDIS_ACTION_RESULTS_GROUP = os.getenv("REDIS_ACTION_RESULTS_GROUP", "action-results-group")
REDIS_WORKFLOW_TRIGGERS_GROUP = os.getenv("REDIS_WORKFLOW_TRIGGERS_GROUP", "workflow-triggers-group")
REDIS_WORKFLOW_CONTROL = os.getenv("REDIS_WORKFLOW_CONTROL", "workflow-control")
REDIS_WORKFLOW_CONTROL_GROUP = os.getenv("REDIS_WORKFLOW_CONTROL_GROUP", "workflow-control-group")
# Overrides the environment variables for docker-compose and docker commands on the docker machine at 'DOCKER_HOST'
# See: https://docs.docker.com/compose/reference/envvars/ for more information.
# DOCKER_HOST = os.environ.get("DOCKER_HOST", "tcp://ip_of_docker_swarm_manager:2376")
# DOCKER_HOST = os.environ.get("DOCKER_HOST", "unix:///var/run/docker.sock")
# DOCKER_TLS_VERIFY = os.environ.get("DOCKER_TLS_VERIFY", "1")
# DOCKER_CERT_PATH = os.environ.get("DOCKER_CERT_PATH", "/Path/to/certs/for/remote/docker/daemon")
def get_int(self, key, default):
return sint(getattr(self, key), default)
def get_float(self, key, default):
return sfloat(getattr(self, key), default)
config = Config()
functions/newworker/docker_helpers.py
-321
View File
@@ -1,321 +0,0 @@
import logging
import os
import re
import json
import copy
import base64
import tarfile
from io import BytesIO
from pathlib import Path
from contextlib import contextmanager, asynccontextmanager
import aiodocker
from aiodocker.utils import clean_map
from aiodocker.exceptions import DockerError
import docker
from docker.models.services import _get_create_service_kwargs
from docker.types.services import ServiceMode, Resources, EndpointSpec, RestartPolicy, SecretReference
from compose.cli.command import get_project as get_compose_project
from compose.utils import timeparse, parse_bytes
from compose.config.environment import Environment
from config import config
from helpers import sint, sfloat
logger = logging.getLogger("UMPIRE")
class DockerBuildError(Exception):
pass
# TODO: Clean a lot of this up and rectify the inconsistencies between the different docker libraries
class ServiceKwargs:
@classmethod
def configure(cls, image, service, secrets=None, mounts=None, **kwargs):
self = ServiceKwargs()
options = service.options
deploy_opts = options.get("deploy", {})
prefs = deploy_opts.get("placement", {}).get("preferences", {})
# Map compose options to service options
self.image = image
self.constraints = deploy_opts.get("placement", {}).get("constraints")
self.preferences = [kv for pref in prefs for kv in pref.items()]
self.container_labels = options.get("labels")
self.endpoint_spec = EndpointSpec(deploy_opts.get("endpoint_mode"),
{p.published: p.target for p in options.get("ports", [])})
self.env = options.get("environment", None)
self.hostname = options.get("hostname")
self.isolation = options.get("isolation")
self.labels = {k: v for k, v in (kv.split('=') for kv in deploy_opts.get("labels", []))}
self.log_driver = options.get("logging", {}).get("driver")
self.log_driver_options = options.get("logging", {}).get("options")
self.mode = ServiceMode(deploy_opts.get("mode", "replicated"), deploy_opts.get("replicas", 1))
self.networks = [config.SWARM_NETWORK] # Similar to mounts. I don't see the use case but see the issues
resource_opts = deploy_opts.get("resources", {})
if resource_opts:
# Unpack any generic_resources defined i.e. gpus and such
reservation_opts = resource_opts.get("reservations", {})
generic_resources = {}
for generic_resource in reservation_opts.get("generic_resources", {}):
discrete_resource_spec = generic_resource["discrete_resource_spec"]
generic_resources[discrete_resource_spec["kind"]] = discrete_resource_spec["value"]
cpu_limit = sfloat(resource_opts.get("limits", {}).get("cpus"), 0)
cpu_reservation = sfloat(reservation_opts.get("cpus"), 0)
nano_cpu_limit = sint(cpu_limit * 1e9, 0) if cpu_limit is not None else None
nano_cpu_reservation = sint(cpu_reservation * 1e9, 0) if cpu_reservation is not None else None
self.resources = Resources(cpu_limit=nano_cpu_limit,
mem_limit=parse_bytes(resource_opts.get("limits", {}).get("memory", '')),
cpu_reservation=nano_cpu_reservation,
mem_reservation=parse_bytes(reservation_opts.get("memory", '')),
generic_resources=generic_resources)
restart_opts = deploy_opts.get("restart_policy", {})
if restart_opts:
# Parse the restart policy
delay = timeparse(restart_opts.get("delay", "0s"))
window = timeparse(restart_opts.get("restart_opts", "0s"))
self.restart_policy = RestartPolicy(condition=restart_opts.get("condition", ),
delay=delay,
max_attempts=sint(restart_opts.get("max_attempts", 0), 0),
window=window)
self.secrets = secrets
self.mounts = mounts
# Grab any key word arguments that may have been given
[setattr(self, k, v) for k, v in kwargs.items() if hasattr(self, k)]
service_kwargs = _get_create_service_kwargs('create', copy.copy(self.__dict__))
# This is needed because aiodocker assumes the Env is a dictionary for some reason...
if self.env is not None:
service_kwargs["task_template"]["ContainerSpec"]["Env"] = self.env
return service_kwargs
async def create_secret(client, name, data):
data = base64.b64encode(data)
data = data.decode("ascii")
body = {"Data": data, "Name": name}
headers = {"Content-Type": "application/json"}
resp = await client._query("secrets/create", "POST", data=json.dumps(body), headers=headers)
return await resp.json()
async def update_service(client, service_id, version, *, image=None, rollback=None, mode=None):
if image is None and rollback is False:
raise ValueError("You need to specify an image.")
inspect_service = await client.services.inspect(service_id)
spec = inspect_service["Spec"]
if mode is not None:
spec["Mode"] = mode
if image is not None:
spec["TaskTemplate"]["ContainerSpec"]["Image"] = image
params = {"version": version}
if rollback is True:
params["rollback"] = "previous"
data = json.dumps(clean_map(spec))
await client._query_json(
"services/{service_id}/update".format(service_id=service_id),
method="POST",
data=data,
params=params,
)
return True
async def get_secret(client: aiodocker.Docker, secret_id):
resp = await client._query(f"secrets/{secret_id}")
print(resp)
print(resp)
return await resp.json()
async def delete_secret(client: aiodocker.Docker, secret_id):
await client._query(f"secrets/{secret_id}", "DELETE")
async def get_nodes(client: aiodocker.Docker):
resp = await client._query("nodes")
return await resp.json()
async def get_tasks(client: aiodocker.Docker, params):
resp = await client._query("tasks" + '?' + params)
return await resp.json()
def normalize_name(name, delimiter=''):
""" Super arbitrary naming convention for docker images/services... """
return re.sub(r'[^-_a-z0-9]', delimiter, name.lower())
def get_project(path):
project = get_compose_project(path, environment=load_docker_env(), project_name=config.APP_PREFIX)
project.path = path # we'll add this in to refresh the project later
return project
def load_docker_env():
# TODO: remove this since it is likely no longer needed
environment = os.environ
# environment.update({key: val for key, val in config["DOCKER_ENV"].items()})
return Environment(environment)
async def get_service(docker_client, service_id):
try:
s = await docker_client.services.inspect(service_id)
return {'id': s["ID"], 'version': s['Version']['Index']}
except DockerError:
return {}
async def remove_service(docker_client, service):
try:
return await docker_client.services.delete(service)
except DockerError:
logger.error(f"Could not delete {service}.")
return False
async def get_replicas(docker_client, service):
"""
Gets the running and desired replica counts for the given service ID
:param service: The docker id of the service
:return: a dictionary giving the number of "running" and "desired" replicas
"""
tasks = await docker_client.tasks.list(filters={"service": [service]})
desired = sum([t["DesiredState"] == "running" for t in tasks])
running = sum([t["Status"]["State"] == "running" for t in tasks])
return {"running": running, "desired": desired}
async def get_containers(docker_client, service, short_ids=False):
"""
Gets the running containers the given service ID
:param service: The docker id of the service
:return: a set of the running containers
"""
def get_container_id(task_spec):
return task_spec["Status"]["ContainerStatus"]["ContainerID"]
def get_state(task_spec):
return task_spec["Status"]["State"]
def has_container(task_spec):
return task_spec["Status"].get("ContainerStatus") is not None
tasks = await docker_client.tasks.list(filters={"service": [service]})
if short_ids:
return set(get_container_id(t)[:12] for t in tasks if get_state(t) == "running" and has_container(t))
return set(get_container_id(t) for t in tasks if get_state(t) == "running" and has_container(t))
async def load_secrets(docker_client, project):
service = project.services[0]
secret_references = []
for service_secret in service.secrets:
secret = service_secret["secret"]
filename = service_secret.get("file", secret.source)
# Compose doesn't parse external secrets so we'll assume there is one and build if it doesn't exist
try:
secret_id = await get_secret(docker_client, secret.source)
except (AttributeError, DockerError):
with open(filename, 'rb') as fp:
data = fp.read()
secret_id = (await create_secret(docker_client, name=secret.source, data=data)).get("ID")
if secret_id is not None:
secret_references.append(SecretReference(secret_id=secret_id, secret_name=secret.source,
uid=secret.uid, gid=secret.gid, mode=secret.mode))
return secret_references
def connect_to_docker():
client = docker.from_env(environment=load_docker_env())
try:
if client.ping():
logger.debug(f"Connected to Docker Engine: v{client.version()['Version']}")
return client
except docker.errors.APIError as e:
logger.error(f"Docker API error during connect: {e}")
@asynccontextmanager
async def connect_to_aiodocker():
client = aiodocker.Docker()
try:
if (await client._query("_ping")).status == 200:
resp = await client._query("version")
version = (await resp.json())["Version"]
logger.debug(f"Connected to Docker Engine: v{version}")
yield client
finally:
await client.close()
logger.info("Docker connection closed.")
@contextmanager
def docker_context(path, dirs=None):
"""
Tars and compresses the given docker context in memory. Useful for sending contexts to `docker build` commands.
:param path: str or pathlib.Path object representing the path of the context
:param dirs: white list of directories under path to grab
:return: an in memory tar of the context
"""
if not isinstance(path, Path):
try:
path = Path(path)
except (ValueError, NotImplementedError):
logger.exception(f"Error accessing path: \"{path}\"")
return
fileobj = BytesIO()
tar = tarfile.open(fileobj=fileobj, mode="w")
# If a list of subdirectories is listed, only grab them
if dirs is not None:
for d in dirs:
tar.add(path / d, arcname=d)
else:
tar.add(path, arcname='')
tar.close()
try:
fileobj.seek(0) # must go back to start of file after tarfile writes to it
yield fileobj
finally:
fileobj.close()
async def stream_docker_log(log_stream):
async for line in log_stream:
if "stream" in line and line["stream"].strip():
print(line["stream"].strip())
logger.debug(line["stream"].strip())
elif "status" in line:
print(line["status"].strip())
logger.debug(line["status"].strip())
elif "error" in line:
print(line["error"].strip())
logger.error(line["error"].strip())
raise DockerBuildError
else:
print(line)
functions/newworker/helpers.py
-101
View File
@@ -1,101 +0,0 @@
import logging
from config import config
import aiohttp
import requests
from message_types import(message_dumps, NodeStatusMessage, WorkflowStatusMessage,
StatusEnum, JSONPatch, JSONPatchOps)
logger = logging.getLogger("WALKOFF")
HEX_CHARS = 'abcdefABCDEF0123456789'
UUID_GLOB = "-".join((f"[{HEX_CHARS}]" * i for i in (8, 4, 4, 4, 12)))
UUID_REGEX = "[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}"
def sint(value, default):
if not isinstance(default, int):
raise TypeError("Default value must be of integer type")
try:
return int(value)
except (TypeError, ValueError):
return default
def sfloat(value, default):
if not isinstance(default, int):
raise TypeError("Default value must be of float type")
try:
return float(value)
except (TypeError, ValueError):
return default
async def get_walkoff_auth_header(session, token=None, timeout=5*60):
url = config.API_GATEWAY_URI.rstrip('/') + '/api'
# TODO: make this secure and don't use default admin user
if token is None:
async with session.post(url + "/auth", json={"username": config.WALKOFF_USERNAME,
"password": config.WALKOFF_PASSWORD}, timeout=timeout) as resp:
resp_json = await resp.json()
#token = resp_json["refresh_token"]
token = "refresh"
logger.debug("Successfully logged into WALKOFF")
headers = {"Authorization": f"Bearer {token}"}
async with session.post(url + "/auth/refresh", headers=headers, timeout=timeout) as resp:
resp_json = await resp.json()
#access_token = resp_json["access_token"]
access_token = "access"
logger.debug("Successfully refreshed WALKOFF JWT")
return {"Authorization": f"Bearer {access_token}"}, token
def make_patch(message, root, op, value_only=False, white_list=None, black_list=None):
if white_list is None and black_list is None:
raise ValueError("Either white_list or black_list must be provided")
if white_list is not None and black_list is not None:
raise ValueError("Either white_list or black_list must be provided, not both")
# convert blacklist to whitelist and grab those attrs from the message
white_list = set(message.__slots__).difference(black_list) if black_list is not None else white_list
if value_only and len(white_list) != 1:
raise ValueError("value_only can only be set if a single key is in white_list")
if value_only:
(key,) = white_list
values = getattr(message, key)
else:
values = {k: getattr(message, k) for k in message.__slots__ if k in white_list}
return JSONPatch(op, path=root, value=values)
def get_patches(message):
patches = []
if isinstance(message, NodeStatusMessage):
root = f"/node_statuses/{message.node_id}"
if message.status == StatusEnum.EXECUTING:
patches.append(make_patch(message, root, JSONPatchOps.ADD, black_list={"result", "completed_at"}))
else:
patches.append(make_patch(message, root, JSONPatchOps.REPLACE, black_list={}))
elif isinstance(message, WorkflowStatusMessage):
if message.status == StatusEnum.EXECUTING:
for key in [attr for attr in message.__slots__ if getattr(message, attr)]:
patches.append(make_patch(message, f"/{key}", JSONPatchOps.REPLACE, value_only=True,
white_list={f"{key}"}))
elif message.status == StatusEnum.COMPLETED or message.status == StatusEnum.ABORTED:
patches.append(make_patch(message, f"/status", JSONPatchOps.REPLACE, value_only=True,
white_list={"status"}))
patches.append(make_patch(message, f"/completed_at", JSONPatchOps.REPLACE, value_only=True,
white_list={"completed_at"}))
return patches
functions/newworker/main.py
-583
View File
@@ -1,583 +0,0 @@
import asyncio
import logging
import json
import sys
import os
import signal
import requests
import os
import time
from collections import deque
from inspect import getcoroutinelocals
from google.cloud import pubsub
import aiohttp
import aioredis
from message_types import message_dumps, message_loads, NodeStatusMessage, WorkflowStatusMessage, StatusEnum
from helpers import get_walkoff_auth_header
from redis_helpers import connect_to_redis_pool, xdel, deref_stream_message
from workflow_types import (Node, Action, Condition, Transform, Parameter, Trigger,
ParameterVariant, Workflow, workflow_dumps, workflow_loads, ConditionException)
logging.basicConfig(level=logging.INFO, format="{asctime} - {name} - {levelname}:{message}", style='{')
logger = logging.getLogger("WORKER")
# logging.getLogger("asyncio").setLevel(logging.DEBUG)
# logger.setLevel(logging.DEBUG)
CONTAINER_ID = ""#os.getenv("HOSTNAME")
APIKEY = ""#os.getenv("FUNCTION_APIKEY")
# FIXME
#apiurl = "http://localhost:5001"
apiurl = "https://shuffler.io"
class Worker:
def __init__(self, workflow: Workflow = None, start_action: str = None, redis: aioredis.Redis = None,
session: aiohttp.ClientSession = None):
self.workflow = workflow
self.start_action = start_action if start_action is not None else self.workflow.start
self.results_stream = f"{workflow.execution_id}:results"
self.parallel_accumulator = {}
self.accumulator = {}
self.parallel_in_process = {}
self.in_process = {}
self.redis = redis
self.streams = set()
self.scheduling_tasks = set()
self.results_getter_task = None
self.parallel_tasks = set()
self.workflow_tasks = set()
self.execution_task = None
self.session = session
self.token = None
self.parent_map = {}
self.cancelled = []
self.results = {}
self.execution_id = ""
self.workflow_id = ""
self.id = ""
self.locations = []
self.project_id = ""
self.authorization = ""
self.start_id = start_action.id if start_action is not None else ""
async def cancel_subgraph(self, node):
"""
Cancels the task related to the current node as well as the tasks related to every child of that node.
Also removes them from the worker's internal in_process queue.
"""
# dependents = self.workflow.get_dependents(node)
cancelled_tasks = set()
self.cancelled.append(node.id)
to_cancel = await self.cancel_helper(node, [node.id])
for task in self.scheduling_tasks:
for _, arg in getcoroutinelocals(task._coro).items():
if isinstance(arg, Node):
if arg.id in to_cancel:
self.in_process.pop(arg.id)
self.accumulator[arg.id] = None
self.cancelled.append(arg.id)
task.cancel()
cancelled_tasks.add(task)
await asyncio.gather(*cancelled_tasks, return_exceptions=True)
# This is a very specific one, that might be fucked up by an action named the same thing.
# Its this way because of a weird translation from Triggers to Actions that didn't
# really work very well
def handle_user_input_node(self, node):
print("Handle user input. Params: %d!" % len(node.parameters))
data = ""
options = ""
actiontypes = []
for parameter in node.parameters:
print("Param: %s" % parameter)
if parameter.name == "alertinfo":
data = parameter.value
elif parameter.name == "options":
options = parameter.value
elif parameter.name == "type":
actiontypes = parameter.value.split(",")
print("Data: ", data)
print("Options: ", options)
print("Types: ", actiontypes)
executed = False
headers = {
"Authorization": "Bearer %s" % APIKEY,
"Content-Type": "application/json",
}
for actiontype in actiontypes:
if actiontype == "email":
print("SEND EMAIL!")
#apiurl = "http://localhost:5001"
mailurl = "%s/functions/sendmail" % apiurl
data = {
"targets": ["frikky@shuffler.io"],
"body": data,
"subject": "Shuffle alert requires input!",
"type": "User input",
"sender_company": "Shuffle",
"reference_execution": self.execution_id,
"workflow_id": self.workflow_id,
"execution_type": options,
"start": node.id,
}
# Add it to actionResult here because of start time!
params = self.dereference_params_pubsub(node)
ret = requests.post(mailurl, headers=headers, json=data)
logger.debug("Ret: %s" % ret.text)
logger.debug("Status: %d" % ret.status_code)
if ret.status_code == 200 or ret.status_code == 201:
executed = True
elif actiontype.lower() == "sms":
print("Handle SMS!")
executed = True
if executed:
actionurl = "%s/api/v1/streams" % apiurl
action = {
"name": node.name,
"app_name": node.app_name,
"app_version": node.app_version,
"label": node.label,
"environment": node.environment,
"id": node.id,
}
action_result = {
"action": action,
"authorization": self.authorization,
"execution_id": self.execution_id,
"result": "",
"started_at": int(time.time()),
"status": "WAITING",
}
actionret = requests.post(actionurl, headers=headers, json=action_result)
logger.debug("Actionret: %d", actionret.status_code)
logger.debug("Actionret: %s", actionret.text)
print("SHOULD KILL THE EXECUTION (stop this branch)!")
def execute_workflow_pubsub(self):
"""
Do a simple BFS to visit and schedule each node in the workflow. We assume every node will run and thus preemptively schedule them all. We will clean up any nodes that will not run due to conditions or triggers
"""
visited = {self.start_action}
queue = deque([self.start_action])
self.scheduling_tasks = set()
while queue:
node = queue.pop()
logger.debug("NODE INFO: %s, %s, %s, %s" % (node.name, node.app_name, node.app_version, node.label))
parents = {n.id: n for n in self.workflow.predecessors(node)} if node is not self.start_action and node.id is not self.workflow.start else {}
children = {n.id: n for n in self.workflow.successors(node)}
for parent_id in parents:
if node.id not in self.parent_map.keys():
self.parent_map[node.id] = 1
else:
self.parent_map[node.id] = self.parent_map[node.id] + 1
self.in_process[node.id] = node
if isinstance(node, Action):
node.execution_id = self.workflow.execution_id # the app needs this as a key for the redis queue
# Custom for trigger actions
if node.name == "User Input" and node.app_name == "User Input":
logger.info("Handling user input!")
# Skipping new nodes
if self.start_id != node.id:
self.handle_user_input_node(node)
break
else:
logger.info("Skipping user input as its start node!")
print("NAME: %s, ENV: %s, LABEL" % (node.name, node.environment))
if node.environment == "cloud":
self.scheduling_tasks.add(self.schedule_node_pubsub(node, parents, children))
print("EXIT NAME: %s, ENV: %s, LABEL" % (node.name, node.environment))
for child in sorted(children.values(), reverse=True):
if child not in visited:
queue.appendleft(child)
visited.add(child)
# Checks whether all actions are finished
finished = self.get_action_results_pubsub()
if finished:
print("Got finished and will return!")
break
def dereference_params_pubsub(self, action: Action):
param_ret = []
global_vars = {}
print(action.parameters)
for param in action.parameters:
data = {"value": param.value, "name": param.name, "action_field": param.action_field, "variant": "STATIC_VALUE"}
if param.variant == ParameterVariant.STATIC_VALUE:
data["variant"] = "STATIC_VALUE"
elif param.variant == ParameterVariant.ACTION_RESULT:
data["variant"] = "ACTION_RESULT"
elif param.variant == ParameterVariant.WORKFLOW_VARIABLE:
data["variant"] = "WORKFLOW_VARIABLE"
elif param.variant == ParameterVariant.GLOBAL:
data["variant"] = "GLOBAL"
else:
logger.error(f"Unable to dereference parameter:{param} for action:{action}")
break
param_ret.append(data)
return param_ret
def abort(self):
logger.info("ABORTING %s BECAUSE OF ERROR WITH FUNCTION EXECUTION" % self.execution_id)
url = f"{apiurl}/api/v1/workflows/{self.workflow_id}/executions/{self.execution_id}/abort"
headers = {
"Content-Type": "application/json",
"Accept": "application/json",
"Authorization": f"Bearer {APIKEY}"
}
ret = requests.get(url, headers=headers, timeout=5)
logger.info("Aborted with status: %d and text:\n%s" % (ret.status_code, ret.text))
sys.exit(0)
def schedule_node_pubsub(self, node, parents, children):
""" Waits until all dependencies of an action are met and then schedules the action """
logger.info(f"Scheduling node {node.id} ({node.name})...")
logger.info(self.accumulator)
while not all(parent.id in self.accumulator for parent in parents.values()):
time.sleep(1)
#await asyncio.sleep(0)
logger.info(f"Node {node.id} ({node.name}) ready to execute.")
# node has more than one parent, check if both parent nodes have been cancelled
if len(parents) > 1:
count = 0
for parent in parents:
if parent in self.cancelled:
count = count + 1
if count == self.parent_map[node.id]:
self.cancel_subgraph(node)
print(type(node))
if isinstance(node, Action):
print("NODE: %s" % node)
params = self.dereference_params_pubsub(node)
print("PARAMS: %s" % params)
# Added authorization to send to function
message = {
"parameters": params,
"execution_id": self.execution_id,
"authorization": self.authorization,
"node_project": self.project_id,
"name": node.name,
"app_name": node.app_name,
"app_version": node.app_version,
"id": node.id,
"label": node.name,
}
headers = {
"Content-Type": "application/json",
"Accept": "application/json",
"Authorization": f"Bearer {APIKEY}"
}
# Uses version for production apps, but ID for private apps
functionname = f"{node.app_name}-{node.app_version}"
if not node.sharing:
functionname = f"{node.app_name}-{node.private_id}"
print(f"Functionname (pre): {functionname}")
functionname = functionname.replace("_", "-")
functionname = functionname.replace(":", "-")
functionname = functionname.replace(".", "-")
functionname = functionname.replace(" ", "-")
print(f"Functionname (post): {functionname}")
logger.info(self.locations)
logger.info(self.project_id)
for location in self.locations:
url = f"https://{location}-{self.project_id}.cloudfunctions.net/{functionname}"
#print(message)
try:
ret = requests.post(url, headers=headers, json=message)
# If any error at all, just quit the entire thing (abort)
if ret.status_code == 500 or ret.status_code == 401:
logger.info("Status: %d. There is an error with ret when starting %s. Should cancel execution and exit. RAW: %s" % (ret.status_code, url, ret.text))
self.abort()
except requests.exceptions.ReadTimeout as e:
logger.debug(e)
logger.info("There is an error with ret (readtimeout). Should cancel execution and exit.")
self.abort()
except requests.exceptions.ConnectionError as e:
logger.debug(e)
logger.info("There is an error with ret (connectionerror). Should cancel execution and exit.")
self.abort()
#logger.debug(ret.text)
logger.debug(ret.status_code)
# FIXME - only in one location, e.g. eu-west?
break
group = f"{node.app_name}:{node.app_version}"
stream = f"{node.execution_id}:{group}"
logger.info(f"Scheduled {node}")
def get_action_results_pubsub(self):
""" Continuously monitors the results queue until all scheduled actions have been completed """
results_stream = f"{self.workflow.execution_id}:results"
# 1. Get the results for the workflowexecution. POST with authorization and ID should do the trick
# 2. Check whether the whole thing is still executing
# 3. Check whether the status of self.in_process is updated, if so, remove it from in progress
# 4. Schedule the next nodes somehow
print(len(self.in_process), len(self.parallel_in_process))
print(self.in_process, len(self.parallel_in_process))
url = f"{apiurl}/api/v1/streams/results"
#if self.project_id != "":
# url = f"https://{self.project_id}.appspot.com/api/v1/streams/results"
#
headers = {"Content-Type": "application/json"}
# Uses workflow specific authorization generated for priviliged access
message = {"authorization": self.authorization, "execution_id": self.execution_id}
sleeptime = 2
logger.info(url)
logger.info(f"Waiting {sleeptime} seconds for new updates in the nodestream...")
while len(self.in_process) > 0 or len(self.parallel_in_process) > 0:
# Ask for all nodes, and check every single one that's in progress
print("Items in process: %s" % self.in_process)
ret = requests.post(url, headers=headers, json=message)
if ret.status_code != 200:
logger.exception("Something went wrong getting workflow status for %s with auth %s. Raw: %s. Status: %d" % (self.execution_id, self.authorization, ret.text, ret.status_code))
time.sleep(sleeptime)
continue
# PAUSED, AWAITING_DATA, PENDING, COMPLETED, ABORTED, EXECUTING, SUCCESS, FAILURE
# FIXME - have this?
if ret.json()["status"] == "FINISHED" or ret.json()["status"] == "ABORTED" or ret.json()["status"] == "FAILURE":
print("Entire thing is done with status %s - exiting" % ret.json()["status"])
return True
self.results = ret.json()
# FIXME - REMOVE COMMENTS
# FIXME - This might be wrong for multiple reasons
if self.results.get("results") == "" or self.results.get("results") == None:
print("Couldn't find results in results - getting new")
logger.info(self.results)
self.results["results"] = []
#print("IS IT DONE? - RETURNING TRUE")
#return
for node_message in self.results["results"]:
# Ensure that the received NodeStatusMessage is for an action we launched
#print(node_message)
#print(self.in_process)
# FIXME - might be an issue with same kind of node with same ID here
if node_message["action"]["id"] in self.in_process:
if node_message["status"] == "EXECUTING":
logger.info(f"Got EXECUTING result for: {node_message['action']['name']}-{node_message['execution_id']}")
elif node_message["status"] == "WAITING":
# This is just for user-inputted items
logger.info("Should only be here the SECOND time around (after user inputted)!")
logger.info(f"Got WAITING result for: {node_message['action']['name']}-{node_message['execution_id']}. Updating it to SUCCESS now that a user continued.")
self.accumulator[node_message["action"]["id"]] = "SUCCESS"
self.in_process.pop(node_message["action"]["id"], None)
logger.debug("start_id: %s, node.id: %s", self.start_id, node_message["action"]["id"])
if self.start_id == node_message["action"]["id"]:
logger.info("HANDLING USER INPUT AS START NODE - SETTING TO SUCCESS!")
# Check if its the same, then update it to success
headers = {
"Authorization": "Bearer %s" % APIKEY,
"Content-Type": "application/json",
}
# Set it to successful here?
actionurl = "%s/api/v1/streams" % apiurl
action_result = node_message
action_result["status"] = "SUCCESS"
action_result["authorization"] = self.authorization
action_result["completed_at"] = int(time.time())
action_result["result"] = "User clicked continue!"
actionret = requests.post(actionurl, headers=headers, json=action_result)
elif node_message["status"] == "SKIPPED":
# FIXME - handle SKIPPED - these are
logger.info(f"GOT SKIPPEED result for: {node_message['action']['name']}-{node_message['execution_id']}")
elif node_message["status"] == "SUCCESS":
# Adds the data to accumulator with success AND
# removes the successful ones, which breaks the loop
self.accumulator[node_message["action"]["id"]] = node_message["result"]
logger.info(f"Worker received result for: {node_message['action']['name']}-{node_message['execution_id']}: {node_message['result']}")
self.in_process.pop(node_message["action"]["id"], None)
elif node_message["status"] == "FAILURE":
self.accumulator[node_message["action"]["id"]] = node_message["result"]
# FIXME - cancel nodes
#await self.cancel_subgraph(self.workflow.nodes[node_message.node_id]) # kill the children!
logger.info(f"Worker received error \"{node_message['result']}\" for: {node_message['action']['name']}-"
f"{node_message['execution_id']}")
else:
logger.error(f"Unknown message status received: {node_message}")
node_message = None
time.sleep(sleeptime)
return False
def abort(message, workflow_id, execution_id):
logger.info("ABORTING %s BECAUSE OF ERROR WITH FUNCTION STARTUP" % execution_id)
logger.info("Message: %s" % message)
url = f"{apiurl}/api/v1/workflows/{workflow_id}/executions/{execution_id}/abort"
headers = {
"Content-Type": "application/json",
"Accept": "application/json",
"Authorization": f"Bearer {APIKEY}"
}
ret = requests.get(url, headers=headers, timeout=5)
logger.info("Aborted with status: %d and text:\n%s" % (ret.status_code, ret.text))
sys.exit(0)
def run_function(message):
messagedata = message
# Raise exception?
if messagedata["type"] != "workflow":
return f"Wrong type" % e, 500
# Required fields
execution_id = messagedata["execution_id"]
workflow_id = messagedata["workflow_id"]
# FIXME - add exception handler -> abort
workflow = workflow_loads(json.dumps(messagedata["workflow"]))
id = messagedata["workflow"]["id"]
locations = messagedata["locations"]
project_id = messagedata["project_id"]
authorization = messagedata["authorization"]
execution_id = messagedata["execution_id"]
workflow_id = messagedata["workflow_id"]
logger.info("Exec_id: %s, authorization: %s" % (execution_id, authorization))
if execution_id == None:
logger.info("NO EXECUTION ID")
abort("NO EXECUTION ID", workflow_id, execution_id)
if len(locations) <= 0:
logger.info("NO LOCATIONS")
abort("NO LOCATIONS", workflow_id, execution_id)
if not project_id:
logger.info("NO PROJECT_ID")
abort("NO PROJECT_ID", workflow_id, execution_id)
if not authorization:
logger.info("NO AUTHORIZATION")
abort("NO AUTHORIZATION", workflow_id, execution_id)
if not workflow_id:
logger.info("NO workflow_id")
abort("NO WORKFLOW_ID", workflow_id, execution_id)
worker = Worker(workflow)
worker.locations = locations
worker.execution_id = execution_id
worker.id = id
worker.project_id = project_id
worker.authorization = authorization
worker.workflow_id = workflow_id
try:
worker.start_id = messagedata["start"]
logger.debug("Start node is %s!" % messagedata["start"])
except KeyError:
try:
worker.start_id = messagedata["workflow"]["start"]
except KeyError:
pass
logger.info("STARTING EXECUTION TASK FOR %s" % execution_id)
try:
worker.execution_task = worker.execute_workflow_pubsub()
except Exception as e:
logger.error("Execution exception: %s" % e)
abort(e, workflow_id, execution_id)
# def abort(self):
logger.info(worker.execution_task)
return f"OK", 200
def authorization(data, context):
logger.info("JUST STARTED")
# Rofl
import base64
data = base64.b64decode(data['data']).decode('utf-8')
return main(data)
def main(data):
import argparse
LOG_LEVELS = ("debug", "info", "error", "warn", "fatal", "DEBUG", "INFO", "ERROR", "WARN", "FATAL")
parser = argparse.ArgumentParser()
parser.add_argument("--log-level", dest="log_level", choices=LOG_LEVELS, default="DEBUG")
parser.add_argument("--debug", "-d", dest="debug", action="store_true",
help="Enables debug level logging for the umpire as well as asyncio debug mode.")
args = parser.parse_args()
logger.setLevel(args.log_level.upper())
logger.info("STARTED")
if isinstance(data, str):
data = json.loads(data)
return run_function(data)
def test():
# Used for testing
with open("data.json", "r") as tmp:
print(main(tmp.read()))
if __name__ == "__main__":
test()
functions/newworker/message_types.py
-214
View File
@@ -1,214 +0,0 @@
import enum
import json
import datetime
def message_dumps(obj):
return json.dumps(obj, cls=MessageJSONEncoder)
def message_loads(obj):
return json.loads(obj, cls=MessageJSONDecoder)
def message_dump(obj, fp):
return json.dump(obj, fp, cls=MessageJSONEncoder)
def message_load(obj):
return json.load(obj, cls=MessageJSONDecoder)
class MessageJSONDecoder(json.JSONDecoder):
""" A custom decoder for decoding JSON strings to Message types. """
def __init__(self, *args, **kwargs):
json.JSONDecoder.__init__(self, object_hook=self.object_hook, *args, **kwargs)
def object_hook(self, o):
if "result" in o and "app_name" in o:
o["status"] = StatusEnum[o["status"]]
return NodeStatusMessage(**o)
elif "workflow_id" in o and "execution_id" in o:
o["status"] = StatusEnum[o["status"]]
return WorkflowStatusMessage(**o)
elif "trigger_data" in o:
return TriggerMessage(**o)
else:
return o
class MessageJSONEncoder(json.JSONEncoder):
""" A custom encoder for encoding Message types to JSON strings. """
def default(self, o):
if isinstance(o, NodeStatusMessage):
r = {"name": o.name, "node_id": o.node_id, "label": o.label, "app_name": o.app_name,
"execution_id": o.execution_id, "result": o.result, "status": o.status,
"started_at": o.started_at, "completed_at": o.completed_at, "combined_id": o.combined_id,
"parameters": o.parameters}
try:
json.dumps(o.result)
except (TypeError, ValueError):
r["result"] = f"Node returned result of type '{type(o.result)}' which is not JSON serializable."
r["status"] = StatusEnum.FAILURE
finally:
return r
elif isinstance(o, WorkflowStatusMessage):
return {"execution_id": o.execution_id, "workflow_id": o.workflow_id, "name": o.name, "status": o.status,
"started_at": o.started_at, "completed_at": o.completed_at, "user": o.user}
elif isinstance(o, TriggerMessage):
return {"trigger_data": o.trigger_data}
elif isinstance(o, JSONPatch):
if o.op in JSONPatchOps:
return {k: getattr(o, k, None) for k in o.__slots__ if getattr(o, k, None) is not None}
else:
raise ValueError("Improper JSON Patch operation")
elif isinstance(o, StatusEnum):
return o.value
elif isinstance(o, JSONPatchOps):
return o.value.lower()
elif isinstance(o, JSONPatch):
return {k: getattr(o, k, None) for k in o.__slots__ if getattr(o, k, None) is not None}
elif isinstance(o, datetime.datetime):
return str(o)
else:
return o
class JSONPatch:
__slots__ = ("op", "path", "value", "from_")
def __init__(self, op=None, path=None, value=None, from_=None):
self.op = op
self.path = path
self.value = value
self.from_ = from_
class JSONPatchOps(enum.Enum):
TEST = "TEST"
REMOVE = "REMOVE"
ADD = "ADD"
REPLACE = "REPLACE"
MOVE = "MOVE"
COPY = "COPY"
class StatusEnum(enum.Enum):
""" Holds statuses used for Workflow and Action status messages """
PAUSED = "PAUSED" # not currently implemented but may be if we see a use case
AWAITING_DATA = "AWAITING_DATA" # possibly for triggers?
PENDING = "PENDING"
COMPLETED = "COMPLETED"
ABORTED = "ABORTED"
EXECUTING = "EXECUTING"
SUCCESS = "SUCCESS"
FAILURE = "FAILURE"
class WorkflowStatusMessage(object):
""" Class that formats a WorkflowStatusMessage message """
__slots__ = ("execution_id", "workflow_id", "name", "status", "started_at", "completed_at", "user")
def __init__(self, execution_id, workflow_id, name, started_at=None, completed_at=None, status=None, user=None):
self.execution_id = execution_id
self.workflow_id = workflow_id
self.name = name
self.status = status
self.started_at = started_at
self.completed_at = completed_at
self.user = user
@classmethod
def execution_pending(cls, execution_id, workflow_id, name, user=None):
return cls(execution_id, workflow_id, name, status=StatusEnum.PENDING, user=user)
@classmethod
def execution_started(cls, execution_id, workflow_id, name, user=None):
start_time = datetime.datetime.now()
return cls(execution_id, workflow_id, name, started_at=start_time, status=StatusEnum.EXECUTING, user=user)
@classmethod
def execution_completed(cls, execution_id, workflow_id, name, user=None):
end_time = datetime.datetime.now()
return cls(execution_id, workflow_id, name, completed_at=end_time, status=StatusEnum.COMPLETED, user=user)
@classmethod
def execution_aborted(cls, execution_id, workflow_id, name, user=None):
end_time = datetime.datetime.now()
return cls(execution_id, workflow_id, name, completed_at=end_time, status=StatusEnum.ABORTED, user=user)
class NodeStatusMessage(object):
""" Class that formats a NodeStatusMessage message. """
__slots__ = ("name", "node_id", "label", "app_name", "execution_id", "parameters", "combined_id", "result",
"status", "started_at", "completed_at")
def __init__(self, name, node_id, label, app_name, execution_id, combined_id=None, parameters=None, result=None,
status=None, started_at=None, completed_at=None):
self.name = name
self.node_id = node_id
self.label = label
self.app_name = app_name
self.execution_id = execution_id
self.combined_id = combined_id if combined_id is not None else ':'.join((node_id, execution_id))
self.result = result
self.parameters = parameters
self.status = status
self.started_at = started_at
self.completed_at = completed_at
@classmethod
def from_node(cls, node, execution_id, result=None, status=None, started_at=None, completed_at=None, parameters=None):
return cls(node.name, node.id, node.label, node.app_name, execution_id, result=result,
status=status, started_at=started_at, completed_at=completed_at, parameters=parameters)
@classmethod
def pending_from_node(cls, node, execution_id, parameters=None):
return NodeStatusMessage.from_node(node, execution_id, status=StatusEnum.PENDING, parameters=parameters)
@classmethod
def executing_from_node(cls, node, execution_id, parameters=None):
started_at = datetime.datetime.now()
return NodeStatusMessage.from_node(node, execution_id, started_at=started_at, status=StatusEnum.EXECUTING,
parameters=parameters)
@classmethod
def success_from_node(cls, node, execution_id, result, parameters=None):
completed_at = datetime.datetime.now()
return NodeStatusMessage.from_node(node, execution_id, result=result, completed_at=completed_at,
status=StatusEnum.SUCCESS, parameters=parameters)
@classmethod
def failure_from_node(cls, node, execution_id, result, parameters=None):
completed_at = datetime.datetime.now()
return NodeStatusMessage.from_node(node, execution_id, result=result, completed_at=completed_at,
status=StatusEnum.FAILURE, parameters=parameters)
@classmethod
def aborted_from_node(cls, node, execution_id, parameters=None):
completed_at = datetime.datetime.now()
return NodeStatusMessage.from_node(node, execution_id, result=None, completed_at=completed_at,
status=StatusEnum.ABORTED, parameters=parameters)
class TriggerMessage(object):
""" Class that formats a TriggerMessage. """
__slots__ = ("trigger_data",)
def __init__(self, trigger_data):
self.trigger_data = trigger_data
functions/newworker/redis_helpers.py
-40
View File
@@ -1,40 +0,0 @@
import logging
from contextlib import asynccontextmanager
import aioredis
logger = logging.getLogger("WALKOFF")
@asynccontextmanager
async def connect_to_redis_pool(redis_uri) -> aioredis.Redis:
# Redis client bound to pool of connections (auto-reconnecting).
redis = await aioredis.create_redis_pool(redis_uri)
try:
yield redis
finally:
# gracefully close pool
redis.close()
await redis.wait_closed()
logger.info("Redis connection pool closed.")
def deref_stream_message(message):
try:
key, value = message[0][-1].popitem()
stream = message[0][0]
id = message[0][1]
return (key, value), stream, id
except:
logger.exception("Stream message formatted incorrectly.")
def xlen(redis: aioredis.Redis, key):
"""Returns the number of entries inside a stream."""
return redis.execute(b'XLEN', key)
def xdel(redis: aioredis.Redis, stream, id):
""" Deletes id from stream. Returns the number of items deleted. """
return redis.execute(b'XDEL', stream, id)
functions/newworker/requirements.txt
-12
View File
@@ -1,12 +0,0 @@
aiodns
aiodocker
aioredis
aiohttp
cchardet
docker
docker-compose
pyyaml
sqlalchemy
asteval
argparse
google-cloud-pubsub
functions/newworker/workflow_types.py
-548
View File
@@ -1,548 +0,0 @@
import uuid
import json
import enum
import logging
from operator import attrgetter, itemgetter
from collections import namedtuple, deque
from asteval import Interpreter, make_symbol_table
logger = logging.getLogger("WALKOFF")
def workflow_dumps(obj):
return json.dumps(obj, cls=WorkflowJSONEncoder)
def workflow_loads(obj):
return json.loads(obj, cls=WorkflowJSONDecoder)
def workflow_dump(obj, fp):
return json.dump(obj, fp, cls=WorkflowJSONEncoder)
def workflow_load(obj, fp):
return json.load(obj, fp, cls=WorkflowJSONDecoder)
def attrs_equal(self, other):
attr_getters = (attrgetter(attr) for attr in self.__slots__)
return all(attr_getter(self) == attr_getter(other) for attr_getter in attr_getters)
class ConditionException(Exception):
pass
class WorkflowJSONDecoder(json.JSONDecoder):
def __init__(self, *args, **kwargs):
super().__init__(object_hook=self.object_hook, *args, **kwargs)
self.nodes = {}
self.branches = set()
def object_hook(self, o):
if "x" in o and "y" in o:
return Point(**o)
elif "parameters" in o and "priority" in o:
node = Action(**o)
self.nodes[node.id] = node
return node
elif "variant" in o:
try:
o["variant"] = ParameterVariant[o["variant"]]
return Parameter(**o)
except KeyError:
o["variant"] = "STATIC_VALUE"
return Parameter(**o)
elif "source_id" in o and "destination_id" in o:
self.branches.add(Branch(source_id=o["source_id"], destination_id=o["destination_id"], id=o["id"]))
elif "conditional" in o:
node = Condition(**o)
self.nodes[node.id] = node
return node
elif "transform" in o:
node = Transform(**o)
self.nodes[node.id] = node
return node
elif "trigger_schema" in o:
node = Trigger(**o)
self.nodes[node.id] = node
return node
elif "description" in o and "value" in o:
return Variable(**o)
elif "actions" in o and "branches" in o:
branches = {Branch(self.nodes[b.source_id], self.nodes[b.destination_id], b.id) for b in self.branches}
try:
workflow_variables = {var.id: var for var in o["workflow_variables"]}
except:
workflow_variables = {}
if o["workflow_variables"] != None:
for var in o["workflow_variables"]:
workflow_obj = Variable(id=var["id"], name=var["name"], value=var["value"])
workflow_variables[workflow_obj.id] = workflow_obj
start = self.nodes[o["start"]]
o["branches"] = branches
o["workflow_variables"] = workflow_variables
o["start"] = start
return Workflow(**o)
else:
return o
class WorkflowJSONEncoder(json.JSONEncoder):
""" A custom encoder for encoding Workflow types to JSON strings.
Note: JSON encoded strings of our custom objects are lossy...for now.
"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.workflow = {}
def default(self, o):
if isinstance(o, Workflow):
# Unpack the adjacency matrix into edges
branches = [{"source_id": src.id, "destination_id": dst.id} for src, dsts in o.edges.items()
for dst in dsts]
branches.sort(key=itemgetter("source_id", "destination_id"))
actions = [action for action in o.actions]
triggers = [trigger for trigger in o.triggers]
workflow_variables = list(o.workflow_variables.values())
return {"id": o.id, "execution_id": o.execution_id, "name": o.name, "start": o.start.id,
"actions": actions, "branches": branches,
"triggers": triggers, "workflow_variables": workflow_variables, "is_valid": o.is_valid,
"errors": None}
elif isinstance(o, Action):
position = {"x": o.position.x, "y": o.position.y}
return {"id": o.id, "name": o.name, "app_name": o.app_name, "app_version": o.app_version,
"label": o.label, "position": position, "parameters": o.parameters, "priority": o.priority,
"execution_id": o.execution_id}
elif isinstance(o, Condition):
position = {"x": o.position.x, "y": o.position.y}
return {"id": o.id, "name": o.name, "app_name": o.app_name, "app_version": o.app_version,
"label": o.label, "position": position, "conditional": o.conditional}
elif isinstance(o, Transform):
position = {"x": o.position.x, "y": o.position.y}
return {"id": o.id, "name": o.name, "app_name": o.app_name, "app_version": o.app_version,
"label": o.label, "position": position, "transform": o.transform, "parameter": o.parameter}
elif isinstance(o, Trigger):
position = {"x": o.position.x, "y": o.position.y}
return {"id": o.id, "name": o.name, "app_name": o.app_name, "app_version": o.app_version,
"label": o.label, "position": position, "trigger_schema": o.trigger_schema}
elif isinstance(o, Parameter):
return {"name": o.name, "variant": o.variant, "value": o.value, "id": o.id}
elif isinstance(o, ParameterVariant):
return o.value
elif isinstance(o, Variable):
return {"description": o.description, "id": o.id, "name": o.name, "value": o.value}
else:
return o
Point = namedtuple("Point", ("x", "y"))
Branch = namedtuple("Branch", ("source_id", "destination_id", "id"))
ParentSymbol = namedtuple("ParentSymbol", "result") # used inside conditions to further mask the parent node attrs
ChildSymbol = namedtuple("ChildSymbol", "id") # used inside conditions to further mask the child node attrs
class ParameterVariant(enum.Enum):
STATIC_VALUE = "STATIC_VALUE"
ACTION_RESULT = "ACTION_RESULT"
WORKFLOW_VARIABLE = "WORKFLOW_VARIABLE"
GLOBAL = "GLOBAL"
class Parameter:
__slots__ = ("name", "value", "selection", "variant", "id", "errors", "parallelized", "description", "required", "schema", "action_field", "multiline", "example")
def __init__(self, name, parallelized=False, selection=[], id=None, value=None, variant=None, errors=None, description="", required=False, schema={}, action_field="", multiline=False, example=""):
self.id = id
self.name = name
self.description = description
self.required = required
self.parallelized = parallelized
self.selection = selection
self.value = value
self.variant = variant
self.schema = schema
self.errors = errors
self.action_field = action_field
self.multiline = multiline
self.example = example
def __str__(self):
return f"Parameter-{self.name}:{self.value}"
def __eq__(self, other):
if isinstance(other, Parameter) and self.__slots__ == other.__slots__:
return attrs_equal(self, other)
return False
def __hash__(self):
return hash(id(self))
class Variable:
"""
A lightweight class representing a WALKOFF WorkflowVariable or Global
"""
__slots__ = ("id", "name", "value", "description")
def __init__(self, id, name, value, description=None):
self.id = id
self.name = name
self.value = value
self.description = description
def __eq__(self, other):
if isinstance(other, self.__class__) and self.__slots__ == other.__slots__:
return attrs_equal(self, other)
return False
def __hash__(self):
return hash(id(self))
class Node:
__slots__ = ("id", "name", "app_name", "app_version", "label", "position", "priority", "errors", "is_valid", "parameters")
def __init__(self, name, position: Point, label, app_name, app_version, parameters=[], id=None, errors=None, is_valid=True, environment="cloud"):
self.id = id if id is not None else str(uuid.uuid4())
self.is_valid = is_valid # ToDo: Is this neccessary?
self.name = name
self.environment = environment
self.app_name = app_name
self.app_version = app_version
self.label = label
self.parameters = parameters
self.position = position
self.errors = errors if errors is not None else []
if hasattr(self, "priority"):
msg = f"Call super().__init__() prior to setting self.priority in Node subclass {self.__class__.__name__}"
logger.warning(msg)
else:
self.priority = 3 # initialize this to mid level for non-Action node types
def __repr__(self):
return f"Node-{self.id}"
def __str__(self):
return f"Node-{self.label}"
def __gt__(self, other):
return self.priority > other.priority
def __eq__(self, other):
if isinstance(other, self.__class__) and self.__slots__ == other.__slots__:
return attrs_equal(self, other)
return False
def __hash__(self):
return hash(id(self))
class Action(Node):
__slots__ = ("parameters", "execution_id", "parallelized", "environment", "authentication", "sharing", "private_id")
def __init__(self, name, position, app_name, app_version, label, priority, environment, sharing=False, private_id="", verified=False, parallelized=False, parameters=None, id=None, execution_id=None, errors=None, is_valid=None, authentication=[], app_id="", **kwargs):
super().__init__(name, position, label, app_name, app_version, id=id, errors=errors, is_valid=is_valid)
self.parameters = parameters if parameters is not None else list()
self.parallelized = parallelized
self.priority = priority
self.execution_id = execution_id
self.authentication = authentication
self.sharing = sharing
self.private_id = private_id
def __str__(self):
return f"Action: {self.label}::{self.id}"
def __repr__(self):
return f"Action: {self.label}::{self.id}"
def __eq__(self, other):
if isinstance(other, self.__class__) and self.__slots__ == other.__slots__:
return attrs_equal(self, other)
return False
def __hash__(self):
return hash(id(self))
class Condition(Node):
__slots__ = ("conditional",)
def __init__(self, name, position: Point, app_name, app_version, label, conditional, id=None, errors=None,
is_valid=None):
super().__init__(name, position, label, app_name, app_version, id, errors, is_valid)
self.conditional = conditional
self.priority = 3 # Conditions have a fixed, mid valued priority
def __str__(self):
return f"Condition: {self.label}::{self.id}"
def __repr__(self):
return f"Condition: {self.label}::{self.id}"
def __eq__(self, other):
if isinstance(other, self.__class__) and self.__slots__ == other.__slots__:
return attrs_equal(self, other)
return False
def __hash__(self):
return hash(id(self))
@staticmethod
def format_node_names(nodes):
# We need to format space delimited names into underscore delimited names
names_to_modify = {node.label for node in nodes.values() if node.label.count(' ') > 0}
formatted_nodes = {}
for node in nodes.values():
formatted_name = node.label.strip().replace(' ', '_')
if formatted_name in names_to_modify: # we have to check for a name conflict as described above
logger.error(f"Error processing condition. {node.label} or {formatted_name} must be renamed.")
formatted_nodes[formatted_name] = node
return formatted_nodes
def __call__(self, parents, children, accumulator) -> str:
parent_symbols = {k: ParentSymbol(accumulator[v.id]) for k, v in self.format_node_names(parents).items()}
children_symbols = {k: ChildSymbol(v.id) for k, v in self.format_node_names(children).items()}
syms = make_symbol_table(use_numpy=False, **parent_symbols, **children_symbols)
aeval = Interpreter(usersyms=syms, no_for=True, no_while=True, no_try=True, no_functiondef=True, no_ifexp=True,
no_listcomp=True, no_augassign=True, no_assert=True, no_delete=True, no_raise=True,
no_print=True, use_numpy=False, builtins_readonly=True,
readonly_symbols=children_symbols.keys())
aeval(self.conditional)
child_id = getattr(aeval.symtable.get("selected_node", None), "id", None)
if len(aeval.error) > 0:
raise ConditionException
return child_id
class Trigger(Node):
__slots__ = ("trigger_schema",)
def __init__(self, name, position: Point, app_name, app_version, label, trigger_schema, id=None, errors=None,
is_valid=None):
super().__init__(name, position, label, app_name, app_version, id, errors, is_valid)
self.trigger_schema = trigger_schema
def __str__(self):
return f"Trigger: {self.label}::{self.id}"
def __repr__(self):
return f"Trigger: {self.label}::{self.id}"
def __eq__(self, other):
if isinstance(other, self.__class__) and self.__slots__ == other.__slots__:
return attrs_equal(self, other)
return False
def __hash__(self):
return hash(id(self))
def __call__(self, data):
""" A trigger simply echos the data it was given """
result = data.trigger_data
logger.debug(f"Executed {self.name}-{self.id} with result: {result}")
return result
class Transform(Node):
__slots__ = ("transform", "parameter")
def __init__(self, name, position: Point, app_name, app_version, label, transform, parameter=None, id=None,
errors=None, is_valid=None):
super().__init__(name, position, label, app_name, app_version, id, errors, is_valid)
self.transform = transform.lower()
self.parameter = parameter
self.priority = 3 # Transforms have a fixed, mid valued priority
def __str__(self):
return f"Transform: {self.label}::{self.id}"
def __repr__(self):
return f"Transform: {self.label}::{self.id}"
def __eq__(self, other):
if isinstance(other, self.__class__) and self.__slots__ == other.__slots__:
return attrs_equal(self, other)
return False
def __hash__(self):
return hash(id(self))
def __call__(self, data):
""" Execute an action and ship its result """
logger.debug(f"Attempting execution of: {self.name}-{self.id}")
transform = f"_{self.__class__.__name__}__{self.transform}"
if hasattr(self, transform):
if self.parameter is None:
result = getattr(self, transform)(data=data)
else:
result = getattr(self, transform)(self.parameter, data=data)
logger.debug(f"Executed {self.name}-{self.id} with result: {result}")
return result
else:
logger.error(f"{self.__class__.__name__} has no method {self.transform}")
# TODO: add JSON to CSV parsing and vice versa.
def __get_value_at_index(self, index, data=None):
return data[index]
def __get_value_at_key(self, key, data=None):
return data[key]
def __split_string_to_array(self, delimiter=' ', data=None):
return data.split(delimiter)
class DiGraph:
__slots__ = ("nodes", "edges", "rev_adjacency")
def __init__(self, nodes, edges):
self.nodes = {}
self.add_nodes(nodes)
self.edges = {node: set() for node in self.nodes.values()}
self.rev_adjacency = {} # all edges inverted for quickly getting parents of a node
self.add_edges(edges)
def __eq__(self, other):
if isinstance(other, self.__class__) and self.__slots__ == other.__slots__:
return attrs_equal(self, other)
return False
def __hash__(self):
return hash(id(self))
def add_edges(self, edges):
try:
iter(edges) # check we got an iterable
if callable(getattr(edges, "items", None)): # check if it's a dictionary
for src, dest in edges.items():
if src in self.edges:
self.edges[src].add(dest)
else: # This edge introduces new nodes so lets add them
self.nodes[src.id] = src
self.nodes[dest.id] = dest
self.edges[src] = {dest}
if dest in self.rev_adjacency:
self.edges[dest].add(src)
else:
self.edges[dest] = {src}
else: # it's a different iterable
for edge in edges:
if not (isinstance(edge, Branch) or (isinstance(edge, tuple) and not len(edge) == 2)):
raise TypeError # it must be an iterable of (src, dest) edges
src = edge[0]
dest = edge[1]
if src in self.edges:
self.edges[src].add(dest)
else:
self.edges[src] = {dest}
if dest in self.rev_adjacency:
self.rev_adjacency[dest].add(src)
else:
self.rev_adjacency[dest] = {src}
except TypeError:
return
def add_edge(self, src, dest):
self.add_edges({src, dest})
def add_nodes(self, nodes):
self.nodes = {node.id: node for node in nodes}
def add_node(self, node):
return self.add_nodes([node])
def successors(self, node):
return self.edges[node]
def predecessors(self, node):
return self.rev_adjacency[node]
# TODO: Maybe look into pooling nodes/branches and sharing them across a workflow to save memory?
class Workflow(DiGraph):
__slots__ = ("start", "id", "id", "is_valid", "name", "execution_id", "workflow_variables",
"triggers", "actions", "errors", "description", "tags", "owner", "org", "execution_org", "schedules", "sharing")
def __init__(self, name, start, actions: [Action], branches: [Branch], workflow_variables=[],
triggers=[], id=None, execution_id=None,
is_valid=None, errors=None, description=None, tags=None, owner={}, org={}, execution_org={}, schedules=[], sharing="private"):
super().__init__(nodes=[*actions, *triggers], edges=branches)
self.start = start
self.id = id if id is not None else str(uuid.uuid4())
self.is_valid = is_valid if is_valid is not None else self.validate()
self.name = name
self.execution_id = execution_id
self.workflow_variables = workflow_variables if workflow_variables is not None else []
self.triggers = triggers
self.actions = actions
self.errors = errors if errors is not None else []
self.description = description
self.owner = owner
self.org = org
self.execution_org = execution_org
self.schedules=schedules
self.tags = tags if tags is not None else []
def __eq__(self, other):
if isinstance(other, self.__class__) and self.__slots__ == other.__slots__:
return attrs_equal(self, other)
return False
def __hash__(self):
return hash(id(self))
def validate(self):
# TODO: add in workflow validation from old implementation
return True
@staticmethod
def dereference_environment_variables(data):
return {ev["id"]: (ev["name"], ev["value"]) for ev in data.get("environment_variables", [])}
def get_dependents(self, node):
"""
BFS to get all nodes dependent on the current node. This includes the current node.
"""
visited = {node}
queue = deque([node])
while queue:
node = queue.pop()
children = set(self.successors(node))
for child in children:
if child not in visited:
queue.appendleft(child)
visited.add(child)
return visited