2323__maintainer__ = "mundialis GmbH & Co. KG"
2424
2525
26+ import json
27+
2628import requests
2729from cloudevents .conversion import to_binary , to_structured
2830from cloudevents .http import CloudEvent , from_http
2931from flask import request
32+ from requests .auth import HTTPBasicAuth
33+
34+ from actinia_cloudevent_plugin .resources .config import ACTINIA , EVENTRECEIVER
3035
3136
3237def receive_cloud_event ():
@@ -43,64 +48,122 @@ def receive_cloud_event():
4348 return event
4449
4550
46- def cloud_event_to_process_chain (event ) -> str :
47- """Return queue name for process chain of event."""
48- # (Remove ruff-exception, when pc variable used)
49- pc = event .get_data ()["list" ][0 ] # noqa: F841
50- # !! TODO !!: pc to job
51+ def start_actinia_job (event ) -> str :
52+ """Return actinia response for process chain of event."""
53+ pc = event .get_data ()
5154 # NOTE: as standalone app -> consider for queue name creation
52- # HTTP POST pc to actinia-module plugin processing endpoint
53- # # # include an identifier for grouping cloudevents of same actinia process (?)
54- # # # (e.g. new metadata field "queue_name", or within data, or use existign id)
55- # -> actinia core returns resource-url, including resource_id (and queue name)
56- # (queuename = xx_<resource_id>; if configured accordingly within actinia -> each job own queue)
57- # via knative jobsink: start actinia worker (with queue name)
58- # (https://knative.dev/docs/eventing/sinks/job-sink/#usage)
59- # e.g. HTTP POST with queue name
60- # kubectl run curl --image=curlimages/curl --rm=true --restart=Never -ti -- -X POST -v \
61- # -H "content-type: application/json" \
62- # -H "ce-specversion: 1.0" \
63- # -H "ce-source: my/curl/command" \
64- # -H "ce-type: my.demo.event" \
65- # -H "ce-id: 123" \
66- # -d '{"details":"queuename"}' \
67- # http://job-sink.knative-eventing.svc.cluster.local/default/job-sink-logger
68- return "<queue_name>_<resource_id>" # queue name and resource id
69-
70-
71- def send_binary_cloud_event (event , actinia_job , url ):
72- """Return posted binary event with actinia_job."""
73- attributes = {
74- "specversion" : event ["specversion" ],
75- "source" : "/actinia-cloudevent-plugin" ,
76- "type" : "com.mundialis.actinia.process.started" ,
77- "subject" : event ["subject" ],
78- "datacontenttype" : "application/json" ,
79- }
80- data = {"actinia_job" : actinia_job }
8155
82- event = CloudEvent (attributes , data )
83- headers , body = to_binary (event )
84- # send event
85- requests .post (url , headers = headers , data = body )
86-
87- return event
56+ # TODO: Define ce attribute for possible mapset.
57+ # Also in actiniaproject divided by "/" or "."?
58+ project = event .get_attributes ().get ("actiniaproject" )
59+ mapset = None
60+ if "." in project :
61+ project = project .split ("." )[0 ]
62+ mapset = project .split ("." )[1 ]
63+
64+ url = f"{ ACTINIA .processing_base_url } { project }
65+ if not mapset :
66+ # emphemeral processing
67+ if ACTINIA .use_actinia_modules :
68+ url += "processing_export"
69+ else :
70+ url += "processing_async_export"
71+ # persistent processing
72+ elif ACTINIA .use_actinia_modules :
73+ url += f"mapsets/{ mapset }
74+ else :
75+ url += f"mapsets/{ mapset }
76+
77+ postkwargs = dict ()
78+ postkwargs ["headers" ] = {"content-type" : "application/json; charset=utf-8" }
79+ postkwargs ["auth" ] = HTTPBasicAuth (ACTINIA .user , ACTINIA .password )
80+ postkwargs ["data" ] = json .dumps (pc )
81+
82+ resp = requests .post (
83+ url ,
84+ ** postkwargs ,
85+ )
86+
87+ # Part of resp:
88+ # 'message' = 'Resource accepted'
89+ # 'queue' = 'job_queue_resource_id-cddae7bb-b4fa-4249-aec4-2a646946ff36'
90+ # 'resource_id' = 'resource_id-cddae7bb-b4fa-4249-aec4-2a646946ff36'
91+ # 'status' = 'accepted'
92+ # 'urls' = {
93+ # 'resources': [],
94+ # 'status':
95+ # 'http://actinia-dev:8088/api/v3/resources/actinia-gdi/
96+ # resource_id-cddae7bb-b4fa-4249-aec4-2a646946ff36'}
97+
98+ return json .loads (resp .text )
99+
100+
101+ def send_binary_cloud_event (event , queue_name , url ):
102+ """Return posted binary event with actinia_job."""
103+ return send_cloud_event (
104+ mode = "binary" ,
105+ version = event ["specversion" ],
106+ cetype = "com.mundialis.actinia.process.startworker" ,
107+ subject = event ["subject" ],
108+ actiniaqueuename = queue_name ,
109+ url = url ,
110+ )
88111
89112
90113def send_structured_cloud_event (event , actinia_job , url ):
91114 """Return posted structured event with actinia_job."""
115+ # TODO: adjust to queue name
116+ return send_cloud_event (
117+ mode = "structured" ,
118+ version = event ["specversion" ],
119+ cetype = "com.mundialis.actinia.process.started" ,
120+ subject = event ["subject" ],
121+ data = {"actinia_job" : actinia_job },
122+ url = url ,
123+ )
124+
125+
126+ def send_cloud_event (
127+ mode = "binary" ,
128+ version = "1.0" ,
129+ cetype = "com.mundialis.actinia.process.started" ,
130+ subject = "nc_spm_08/PERMANENT" ,
131+ actiniaqueuename = None ,
132+ data = "{}" ,
133+ url = None ,
134+ ):
135+ """Post event and return it."""
136+ if url is None :
137+ url = EVENTRECEIVER .url
138+
92139 attributes = {
93- "specversion" : event [ "specversion" ] ,
140+ "specversion" : version ,
94141 "source" : "/actinia-cloudevent-plugin" ,
95- "type" : "com.mundialis.actinia.process.started" ,
96- "subject" : event [ " subject" ] ,
142+ "type" : cetype ,
143+ "subject" : subject ,
97144 "datacontenttype" : "application/json" ,
145+ "actiniaqueuename" : actiniaqueuename ,
98146 }
99- data = {"actinia_job" : actinia_job }
100147
101148 event = CloudEvent (attributes , data )
102- headers , body = to_structured (event )
103- # send event
149+
150+ # From https://github.com/cloudevents/spec/blob/main/cloudevents/spec.md#message
151+ # A "structured-mode message" is one where the entire event (attributes and data)
152+ # are encoded in the message body, according to a specific event format.
153+ # A "binary-mode message" is one where the event data is stored in the message body,
154+ # and event attributes are stored as part of message metadata.
155+ # Often, binary mode is used when the producer of the CloudEvent wishes to add the
156+ # CloudEvent's metadata to an existing event without impacting the message's body.
157+ # In most cases a CloudEvent encoded as a binary-mode message will not break an
158+ # existing receiver's processing of the event because the message's metadata
159+ # typically allows for extension attributes.
160+ # In other words, a binary formatted CloudEvent would work for both
161+ # a CloudEvents enabled receiver as well as one that is unaware of CloudEvents.
162+ if mode == "binary" :
163+ headers , body = to_binary (event )
164+ else :
165+ headers , body = to_structured (event )
166+
104167 requests .post (url , headers = headers , data = body )
105168
106169 return event
0 commit comments