Skip to content

Commit

Permalink
KOGITO-9158 Use RestWorkItemHandler as HTTP client in Knative custom …
Browse files Browse the repository at this point in the history
…function (#2993)

* KOGITO-9158 Used RestWorkItemHandler as HTTP client in Knative custom function

Signed-off-by: Helber Belmiro <[email protected]>

* KOGITO-9158 Refactoring KnativeTypeHandler.java

Signed-off-by: Helber Belmiro <[email protected]>

* KOGITO-9158 Changed request timeout

Signed-off-by: Helber Belmiro <[email protected]>

* KOGITO-9158 Changed KnativeWorkItemHandler to extend RestWorkItemHandler

Signed-off-by: Helber Belmiro <[email protected]>

* KOGITO-9158 Fixed KnativeWorkItemHandlerProducer static method

Signed-off-by: Helber Belmiro <[email protected]>

---------

Signed-off-by: Helber Belmiro <[email protected]>
  • Loading branch information
hbelmiro authored Aug 11, 2023
1 parent 6c0d5d4 commit db2e6fc
Show file tree
Hide file tree
Showing 35 changed files with 1,253 additions and 986 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,86 @@
*/
package org.kie.kogito.addons.quarkus.knative.serving.deployment.customfunctions;

import java.util.ArrayList;
import java.util.List;
import java.util.function.Supplier;

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 com.github.javaparser.ast.expr.Expression;

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;
import static org.kie.kogito.serverless.workflow.utils.ServerlessWorkflowUtils.runtimeRestApi;

public class KnativeTypeHandler extends WorkItemTypeHandler {

private static final String DEFAULT_REQUEST_TIMEOUT_VALUE = "10000";

@Override
public NodeFactory<?, ?> getActionNode(Workflow workflow, ParserContext context,
RuleFlowNodeContainerFactory<?, ?> embeddedSubProcess, FunctionDefinition functionDef,
FunctionRef functionRef, VariableInfo varInfo) {
validateArgs(functionRef);

WorkItemNodeFactory<?> node = buildWorkItem(embeddedSubProcess, context, varInfo.getInputVar(), varInfo.getOutputVar())
.name(functionDef.getName());

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));
}
}

return addFunctionArgs(workflow,
fillWorkItemHandler(workflow, context, node, functionDef),
functionRef);
}

@Override
protected <T extends RuleFlowNodeContainerFactory<T, ?>> WorkItemNodeFactory<T> fillWorkItemHandler(Workflow workflow,
ParserContext context,
WorkItemNodeFactory<T> node,
FunctionDefinition functionDef) {
protected <T extends RuleFlowNodeContainerFactory<T, ?>> WorkItemNodeFactory<T> fillWorkItemHandler(
Workflow workflow, ParserContext context, WorkItemNodeFactory<T> node, FunctionDefinition functionDef) {
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());
}

Supplier<Expression> requestTimeout = runtimeRestApi(functionDef, "timeout",
context.getContext(), String.class, DEFAULT_REQUEST_TIMEOUT_VALUE);

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)
.workParameter(RestWorkItemHandler.REQUEST_TIMEOUT_IN_MILLIS, requestTimeout)
.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 db2e6fc

Please sign in to comment.