Skip to content

Commit

Permalink
KOGITO-9158 Used RestWorkItemHandler as HTTP client in Knative custom…
Browse files Browse the repository at this point in the history
… function

Signed-off-by: Helber Belmiro <[email protected]>
  • Loading branch information
hbelmiro committed Aug 4, 2023
1 parent d838ba2 commit e4c2808
Show file tree
Hide file tree
Showing 34 changed files with 1,280 additions and 985 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@

import java.net.MalformedURLException;
import java.net.URL;
import java.time.Duration;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
Expand Down Expand Up @@ -82,6 +83,8 @@ public class RestWorkItemHandler implements KogitoWorkItemHandler {
public static final String PATH_PARAM_RESOLVER = "PathParamResolver";
public static final String AUTH_METHOD = "AuthMethod";

public static final String REQUEST_TIMEOUT_IN_MILLIS = "RequestTimeout";

public static final int DEFAULT_PORT = 80;
public static final int DEFAULT_SSL_PORT = 443;

Expand Down Expand Up @@ -174,14 +177,38 @@ public void executeWorkItem(KogitoWorkItem workItem, KogitoWorkItemManager manag
requestDecorators.forEach(d -> d.decorate(workItem, parameters, request));
authDecorators.forEach(d -> d.decorate(workItem, parameters, request));
paramsDecorator.decorate(workItem, parameters, request);
HttpResponse<Buffer> response = method.equals(HttpMethod.POST) || method.equals(HttpMethod.PUT) ? request.sendJsonAndAwait(bodyBuilder.apply(parameters)) : request.sendAndAwait();
Duration requestTimeout = getRequestTimeout(parameters);
HttpResponse<Buffer> response = method.equals(HttpMethod.POST) || method.equals(HttpMethod.PUT)
? sendJson(request, bodyBuilder.apply(parameters), requestTimeout)
: send(request, requestTimeout);
int statusCode = response.statusCode();
if (statusCode < 200 || statusCode >= 300) {
throw new WorkItemExecutionException(Integer.toString(statusCode), "Request for endpoint " + endPoint + " failed with message: " + response.statusMessage());
}
manager.completeWorkItem(workItem.getStringId(), Collections.singletonMap(RESULT, resultHandler.apply(response, targetInfo)));
}

private static HttpResponse<Buffer> sendJson(HttpRequest<Buffer> request, Object body, Duration requestTimeout) {
if (requestTimeout == null) {
return request.sendJsonAndAwait(body);
} else {
return request.sendJson(body).await().atMost(requestTimeout);
}
}

private static HttpResponse<Buffer> send(HttpRequest<Buffer> request, Duration requestTimeout) {
if (requestTimeout == null) {
return request.sendAndAwait();
} else {
return request.send().await().atMost(requestTimeout);
}
}

private static Duration getRequestTimeout(Map<String, Object> parameters) {
Long requestTimeoutInMillis = getParam(parameters, REQUEST_TIMEOUT_IN_MILLIS, Long.class, null);
return requestTimeoutInMillis == null ? null : Duration.ofMillis(requestTimeoutInMillis);
}

private Class<?> getTargetInfo(KogitoWorkItem workItem) {
String varName = ((WorkItemNode) ((WorkItemNodeInstance) workItem.getNodeInstance()).getNode()).getIoSpecification().getOutputMappingBySources().get(RESULT);
if (varName != null) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ public static void createServiceIfNotExists(KubernetesServer k8sServer, String k

@SuppressWarnings("deprecation") // Quarkus LTS 2.13 compatibility
public static void createServiceIfNotExists(KubernetesServer k8sServer, String knativeYaml, String namespace, String serviceName, String remoteServiceUrl) {
if (k8sServer.getClient().services().inNamespace("test").withName(serviceName).get() == null) {
if (k8sServer.getClient().services().inNamespace(namespace).withName(serviceName).get() == null) {
KnativeClient knativeClient = k8sServer.getClient().adapt(KnativeClient.class);

Service service = knativeClient.services()
Expand Down
4 changes: 4 additions & 0 deletions quarkus/addons/knative/serving/deployment/pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,10 @@
<groupId>io.quarkus</groupId>
<artifactId>quarkus-vertx-deployment</artifactId>
</dependency>
<dependency>
<groupId>org.kie.kogito</groupId>
<artifactId>kogito-serverless-workflow-rest-parser</artifactId>
</dependency>
</dependencies>

<build>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -15,29 +15,77 @@
*/
package org.kie.kogito.addons.quarkus.knative.serving.deployment.customfunctions;

import java.util.ArrayList;
import java.util.List;

import org.jbpm.compiler.canonical.descriptors.TaskDescriptor;
import org.jbpm.ruleflow.core.RuleFlowNodeContainerFactory;
import org.jbpm.ruleflow.core.factory.NodeFactory;
import org.jbpm.ruleflow.core.factory.WorkItemNodeFactory;
import org.kie.kogito.addons.quarkus.knative.serving.customfunctions.CloudEventKnativeParamsDecorator;
import org.kie.kogito.addons.quarkus.knative.serving.customfunctions.KnativeWorkItemHandler;
import org.kie.kogito.addons.quarkus.knative.serving.customfunctions.Operation;
import org.kie.kogito.addons.quarkus.knative.serving.customfunctions.PlainJsonKnativeParamsDecorator;
import org.kie.kogito.serverless.workflow.parser.ParserContext;
import org.kie.kogito.serverless.workflow.parser.VariableInfo;
import org.kie.kogito.serverless.workflow.parser.types.WorkItemTypeHandler;
import org.kie.kogito.serverless.workflow.suppliers.ParamsRestBodyBuilderSupplier;
import org.kogito.workitem.rest.RestWorkItemHandler;

import io.serverlessworkflow.api.Workflow;
import io.serverlessworkflow.api.functions.FunctionDefinition;
import io.serverlessworkflow.api.functions.FunctionRef;
import io.vertx.core.http.HttpMethod;

import static org.kie.kogito.addons.quarkus.knative.serving.customfunctions.KnativeWorkItemHandler.PAYLOAD_FIELDS_PROPERTY_NAME;
import static org.kie.kogito.serverless.workflow.parser.FunctionTypeHandlerFactory.trimCustomOperation;

public class KnativeTypeHandler extends WorkItemTypeHandler {

@Override
protected <T extends RuleFlowNodeContainerFactory<T, ?>> WorkItemNodeFactory<T> fillWorkItemHandler(Workflow workflow,
ParserContext context,
WorkItemNodeFactory<T> node,
FunctionDefinition functionDef) {
public NodeFactory<?, ?> getActionNode(Workflow workflow, ParserContext context,
RuleFlowNodeContainerFactory<?, ?> embeddedSubProcess, FunctionDefinition functionDef,
FunctionRef functionRef, VariableInfo varInfo) {
validateArgs(functionRef);
return addFunctionArgs(workflow,
fillWorkItemHandler(buildWorkItem(embeddedSubProcess, context, varInfo.getInputVar(), varInfo.getOutputVar()).name(functionDef.getName()), functionDef, functionRef),
functionRef);
}

@Override
protected <T extends RuleFlowNodeContainerFactory<T, ?>> WorkItemNodeFactory<T> fillWorkItemHandler(
Workflow workflow, ParserContext context, WorkItemNodeFactory<T> node, FunctionDefinition functionDef) {
throw new UnsupportedOperationException("This method must never be invoked.");
}

private <T extends RuleFlowNodeContainerFactory<T, ?>> WorkItemNodeFactory<T> fillWorkItemHandler(
WorkItemNodeFactory<T> node, FunctionDefinition functionDef, FunctionRef functionRef) {
if (functionRef.getArguments() != null && !functionRef.getArguments().isEmpty()) {
List<String> payloadFields = new ArrayList<>();
functionRef.getArguments().fieldNames().forEachRemaining(payloadFields::add);
if (!payloadFields.isEmpty()) {
node.workParameter(PAYLOAD_FIELDS_PROPERTY_NAME, String.join(";", payloadFields));
}
}

if (functionDef.getMetadata() != null) {
functionDef.getMetadata().forEach(node::metaData);
}

return node.workName(KnativeWorkItemHandler.NAME).metaData(KnativeWorkItemHandler.OPERATION_PROPERTY_NAME, trimCustomOperation(functionDef));
Operation operation = Operation.parse(trimCustomOperation(functionDef));

if (operation.isCloudEvent()) {
node.workParameter(RestWorkItemHandler.PARAMS_DECORATOR, CloudEventKnativeParamsDecorator.class.getName());
} else {
node.workParameter(RestWorkItemHandler.PARAMS_DECORATOR, PlainJsonKnativeParamsDecorator.class.getName());
}

return node.workParameter(KnativeWorkItemHandler.SERVICE_PROPERTY_NAME, operation.getService())
.workParameter(KnativeWorkItemHandler.PATH_PROPERTY_NAME, operation.getPath())
.workParameter(RestWorkItemHandler.BODY_BUILDER, new ParamsRestBodyBuilderSupplier())
.workParameter(RestWorkItemHandler.METHOD, HttpMethod.POST)
.metaData(TaskDescriptor.KEY_WORKITEM_TYPE, RestWorkItemHandler.REST_TASK_TYPE)
.workName(KnativeWorkItemHandler.NAME);
}

@Override
Expand Down
136 changes: 136 additions & 0 deletions quarkus/addons/knative/serving/integration-tests/pom.xml
Original file line number Diff line number Diff line change
@@ -0,0 +1,136 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.kie.kogito</groupId>
<artifactId>kogito-addons-quarkus-knative-serving-parent</artifactId>
<version>2.0.0-SNAPSHOT</version>
</parent>

<artifactId>kogito-addons-quarkus-knative-serving-integration-tests</artifactId>
<name>Kogito Add-On Knative Serving - Integration Tests</name>
<description>Knative Serving Kogito Add-On Integration Tests.</description>

<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.kie.kogito</groupId>
<artifactId>kogito-quarkus-bom</artifactId>
<version>${project.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>

<dependencies>
<dependency>
<groupId>io.quarkus</groupId>
<artifactId>quarkus-smallrye-openapi</artifactId>
</dependency>
<dependency>
<groupId>org.kie.kogito</groupId>
<artifactId>kogito-quarkus-serverless-workflow</artifactId>
</dependency>

<dependency>
<groupId>io.quarkus</groupId>
<artifactId>quarkus-resteasy</artifactId>
</dependency>
<dependency>
<groupId>io.quarkus</groupId>
<artifactId>quarkus-resteasy-jackson</artifactId>
</dependency>
<dependency>
<groupId>org.kie.kogito</groupId>
<artifactId>kogito-addons-quarkus-knative-serving</artifactId>
</dependency>
<dependency>
<groupId>org.kie.kogito</groupId>
<artifactId>kogito-addons-quarkus-kubernetes</artifactId>
</dependency>
<dependency>
<groupId>org.kie.kogito</groupId>
<artifactId>kogito-addons-quarkus-fabric8-kubernetes-service-catalog</artifactId>
</dependency>
<dependency>
<groupId>io.quarkus</groupId>
<artifactId>quarkus-kubernetes</artifactId>
</dependency>

<!-- Test -->
<dependency>
<groupId>org.assertj</groupId>
<artifactId>assertj-core</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.quarkus</groupId>
<artifactId>quarkus-junit5</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.rest-assured</groupId>
<artifactId>rest-assured</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.github.tomakehurst</groupId>
<artifactId>wiremock-jre8</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.quarkus</groupId>
<artifactId>quarkus-test-kubernetes-client</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.kie.kogito</groupId>
<artifactId>kogito-addons-quarkus-fabric8-kubernetes-service-catalog-test-utils</artifactId>
<scope>test</scope>
</dependency>
</dependencies>

<build>
<pluginManagement>
<plugins>
<plugin>
<groupId>io.quarkus</groupId>
<artifactId>quarkus-maven-plugin</artifactId>
<configuration>
<noDeps>true</noDeps>
<skip>${skipTests}</skip>
</configuration>
</plugin>
</plugins>
</pluginManagement>
<plugins>
<plugin>
<groupId>io.quarkus</groupId>
<artifactId>quarkus-maven-plugin</artifactId>
<executions>
<execution>
<goals>
<goal>build</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
<profiles>
<profile>
<id>native</id>
<activation>
<property>
<name>native</name>
</property>
</activation>
<properties>
<quarkus.package.type>native</quarkus.package.type>
</properties>
</profile>
</profiles>
</project>
Original file line number Diff line number Diff line change
@@ -0,0 +1,34 @@
{
"id": "arrayKnativeFunction",
"version": "1.0",
"name": "Test Knative function",
"description": "This workflow tests a Knative function",
"start": "invokeFunction",
"functions": [
{
"name": "greet",
"type": "custom",
"operation": "knative:services.v1.serving.knative.dev/default/serverless-workflow-greeting-quarkus?path=/arrayFunction"
}
],
"states": [
{
"name": "invokeFunction",
"type": "operation",
"actions": [
{
"functionRef": {
"refName": "greet",
"arguments": {
"array": [
"Javierito",
"Pepito"
]
}
}
}
],
"end": true
}
]
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,38 @@
{
"id": "cloudEvent",
"version": "1.0",
"name": "Test Knative function",
"description": "This workflow tests a Knative function",
"start": "invokeFunction",
"functions": [
{
"name": "greet",
"type": "custom",
"operation": "knative:services.v1.serving.knative.dev/default/serverless-workflow-greeting-quarkus?asCloudEvent=true&path=/cloud-event"
}
],
"states": [
{
"name": "invokeFunction",
"type": "operation",
"actions": [
{
"functionRef": {
"refName": "greet",
"arguments": {
"specversion": "1.0",
"id": "42",
"source": "https://localhost:8080",
"type": "org.kie.kogito.test",
"data": {
"org": "Acme",
"project": "Kogito"
}
}
}
}
],
"end": true
}
]
}
Loading

0 comments on commit e4c2808

Please sign in to comment.