Get.java

package io.kestra.plugin.kubernetes.kubectl;

import java.io.BufferedWriter;
import java.io.FileWriter;
import java.io.IOException;
import java.net.URI;
import java.time.Duration;
import java.util.AbstractMap;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.slf4j.Logger;

import io.kestra.core.models.annotations.Example;
import io.kestra.core.models.annotations.Metric;
import io.kestra.core.models.annotations.Plugin;
import io.kestra.core.models.executions.metrics.Counter;
import io.kestra.core.models.property.Property;
import io.kestra.core.models.tasks.RunnableTask;
import io.kestra.core.models.tasks.common.FetchType;
import io.kestra.core.runners.RunContext;
import io.kestra.core.serializers.FileSerde;
import io.kestra.plugin.kubernetes.AbstractPod;
import io.kestra.plugin.kubernetes.models.Metadata;
import io.kestra.plugin.kubernetes.models.ResourceStatus;
import io.kestra.plugin.kubernetes.services.PodService;
import io.kestra.plugin.kubernetes.services.ResourceWaitService;

import io.fabric8.kubernetes.client.KubernetesClientException;
import io.fabric8.kubernetes.client.dsl.base.ResourceDefinitionContext;
import io.swagger.v3.oas.annotations.media.Schema;
import jakarta.validation.constraints.NotNull;
import lombok.*;
import lombok.experimental.SuperBuilder;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;

import static io.kestra.core.models.tasks.common.FetchType.NONE;
import io.kestra.core.models.annotations.PluginProperty;

@SuperBuilder
@ToString
@EqualsAndHashCode
@Getter
@NoArgsConstructor
@Plugin(
    examples = {
        @Example(
            title = "Get all pods from Kubernetes with a service account and log the status of the pods.",
            full = true,
            code = """
                id: get_all_pods
                namespace: company.team

                tasks:
                  - id: get
                    type: io.kestra.plugin.kubernetes.kubectl.Get
                    connection:
                      masterUrl: "{{ secret('K8S_MASTER_URL') }}"
                      oauthToken: "{{ secret('K8S_TOKEN') }}"
                    namespace: default
                    resourceType: pods
                    fetchType: FETCH

                  - id: log
                    type: io.kestra.plugin.core.log.Log
                    message: "{{ outputs.get.statusItems }}"
                """
        ),
        @Example(
            title = "Get one deployment named 'my-deployment' from Kubernetes with a service account and log the status of the pod.",
            full = true,
            code = """
                id: get_one_deployment
                namespace: company.team

                tasks:
                  - id: get
                    type: io.kestra.plugin.kubernetes.kubectl.Get
                    connection:
                      masterUrl: "{{ secret('K8S_MASTER_URL') }}"
                      oauthToken: "{{ secret('K8S_TOKEN') }}"
                      trustCerts: true
                    namespace: default
                    resourceType: deployments
                    resourcesNames:
                      - my-deployment
                    fetchType: FETCH_ONE

                  - id: log_status
                    type: io.kestra.plugin.core.log.Log
                    message: "{{ outputs.get.statusItem }}"
                """
        ),
        @Example(
            title = "Get two deployments named my-deployment and my-deployment-2 from Kubernetes and store them in the internal storage.",
            full = true,
            code = """
                id: get_two_deployments
                namespace: company.team

                tasks:
                  - id: get
                    type: io.kestra.plugin.kubernetes.kubectl.Get
                    connection:
                      masterUrl: "{{ secret('K8S_MASTER_URL') }}"
                      oauthToken: "{{ secret('K8S_TOKEN') }}"
                    namespace: default
                    resourceType: deployments
                    resourcesNames:
                      - my-deployment
                      - my-deployment-2
                    fetchType: STORE
                """
        ),
        @Example(
            title = "Get one custom resource named Shirt from Kubernetes.",
            full = true,
            code = """
                id: get_one_custom_resource
                namespace: company.team

                tasks:
                  - id: get
                    type: io.kestra.plugin.kubernetes.kubectl.Get
                    connection:
                      masterUrl: "{{ secret('K8S_MASTER_URL') }}"
                      oauthToken: "{{ secret('K8S_TOKEN') }}"
                    namespace: default
                    resourceType: shirts # could be Shirt
                    apiGroup: stable.example.com
                    apiVersion: v1
                    fetchType: FETCH_ONE
                """
        ),
        @Example(
            title = "Get a custom resource and wait for it to become ready.",
            full = true,
            code = """
                id: get_and_wait_for_custom_resource
                namespace: company.team

                tasks:
                  - id: get
                    type: io.kestra.plugin.kubernetes.kubectl.Get
                    connection:
                      masterUrl: "{{ secret('K8S_MASTER_URL') }}"
                      oauthToken: "{{ secret('K8S_TOKEN') }}"
                    namespace: default
                    resourceType: myresource
                    apiGroup: example.com
                    apiVersion: v1
                    resourcesNames:
                      - my-resource
                    fetchType: FETCH_ONE
                    waitUntilReady: PT10M
                """
        ),
    },
    metrics = {
        @Metric(
            name = "fetch.size",
            type = Counter.TYPE,
            unit = "records",
            description = "The number of rows fetch."
        ),
    }
)
@Schema(
    title = "Fetch Kubernetes resources with optional storage",
    description = "Reads resources of a given kind in a namespace, optionally filtering by name. Supports waiting for readiness, returning data to the flow, or storing results in internal storage depending on fetchType."
)
public class Get extends AbstractPod implements RunnableTask<Get.Output> {

    @Schema(
        title = "Resource kind",
        description = "Kubernetes kind (e.g., Pod, Deployment, Service). Case-insensitive."
    )
    @NotNull
    @PluginProperty(group = "main")
    private Property<String> resourceType;

    @Schema(
        title = "Resource names",
        description = "Optional list of names to fetch. When empty, all resources of the kind in the namespace are returned."
    )
    @PluginProperty(group = "source")
    private Property<List<String>> resourcesNames;

    @Schema(
        title = "API group",
        description = "Group for the resource kind (empty for core resources)."
    )
    @PluginProperty(group = "advanced")
    private Property<String> apiGroup;

    @Schema(
        title = "API version",
        description = "Version for the resource kind. Defaults to v1 when omitted."
    )
    @PluginProperty(group = "advanced")
    private Property<String> apiVersion;

    @Schema(
        title = "Fetch behavior",
        description = "Determines the output: NONE returns only metrics; FETCH returns lists; FETCH_ONE returns a single item; STORE writes to internal storage and returns URI."
    )
    @NotNull
    @Builder.Default
    @PluginProperty(group = "processing")
    protected Property<FetchType> fetchType = Property.ofValue(NONE);

    @Override
    public Output run(RunContext runContext) throws Exception {
        var rNamespace = runContext.render(this.namespace).as(String.class)
            .orElseThrow(() -> new IllegalArgumentException("namespace must be provided and rendered."));
        var rResourceType = runContext.render(this.resourceType).as(String.class)
            .orElseThrow(() -> new IllegalArgumentException("resourceType must be provided and rendered."));
        var rResourcesNames = runContext.render(this.resourcesNames).asList(String.class);
        var rApiGroup = runContext.render(this.apiGroup).as(String.class).orElse("");
        var rApiVersion = runContext.render(this.apiVersion).as(String.class).orElse("v1");
        var rFetchType = runContext.render(this.fetchType).as(FetchType.class).orElse(NONE);
        var rWaitUntilReady = runContext.render(this.waitUntilReady).as(Duration.class).orElse(Duration.ZERO);

        List<Metadata> metadataList = new ArrayList<>();
        List<ResourceStatus> statusList = new ArrayList<>();
        Logger logger = runContext.logger();

        try (var client = PodService.client(runContext, this.getConnection())) {

            var resourceDefinitionContext = new ResourceDefinitionContext.Builder()
                // See: https://kubernetes.io/docs/reference/using-api/api-concepts/#resource-uris
                .withGroup(rApiGroup)
                .withVersion(rApiVersion)
                .withKind(rResourceType)
                .withNamespaced(true) // Assuming resources are namespaced as we take namespace input
                .build();

            if (rResourcesNames.isEmpty()) {
                logger.debug("Fetching all resources of kind '{}' in namespace '{}'", rResourceType, rNamespace);
                var resources = client.genericKubernetesResources(resourceDefinitionContext)
                    .inNamespace(rNamespace)
                    .list()
                    .getItems();

                for (var resource : resources) {
                    if (resource != null && resource.getMetadata() != null) {
                        metadataList.add(Metadata.from(resource.getMetadata()));
                        statusList.add(ResourceStatus.from(resource));
                    }
                }
                logger.info("Fetched {} resource(s) of kind '{}' in namespace '{}'", metadataList.size(), rResourceType, rNamespace);

            } else {
                rResourcesNames.forEach(name ->
                {
                    logger.debug(
                        "Fetching resource of kind '{}' with name '{}' in namespace '{}'",
                        rResourceType, name, rNamespace
                    );

                    var resource = client.genericKubernetesResources(resourceDefinitionContext)
                        .inNamespace(rNamespace)
                        .withName(name)
                        .get();

                    if (resource != null && resource.getMetadata() != null) {
                        // Optionally wait for resource to become ready
                        if (!rWaitUntilReady.isZero()) {
                            runContext.logger().info("Waiting for resource '{}' to become ready (timeout: {})...", name, rWaitUntilReady);
                            resource = ResourceWaitService.waitForReady(
                                client,
                                resourceDefinitionContext,
                                rNamespace,
                                name,
                                rWaitUntilReady,
                                runContext.logger()
                            );
                            runContext.logger().info("Resource '{}' is ready", name);
                        }

                        metadataList.add(Metadata.from(resource.getMetadata()));
                        statusList.add(ResourceStatus.from(resource));
                        logger.info(
                            "Fetched resource of kind '{}' with name '{}' in namespace '{}'",
                            rResourceType, name, rNamespace
                        );
                    } else {
                        logger.warn(
                            "Resource of kind '{}' with name '{}' not found in namespace '{}'",
                            rResourceType, name, rNamespace
                        );
                    }
                }
                );
            }

        } catch (KubernetesClientException e) {
            logger.error("Kubernetes API error while fetching kind '{}' in namespace '{}': {}", rResourceType, rNamespace, e.getMessage(), e);
            throw new Exception("Failed to interact with Kubernetes API: " + e.getMessage(), e);
        } catch (IllegalArgumentException e) {
            logger.error("Configuration error: {}", e.getMessage(), e);
            throw e;
        }

        Output output;

        int fetchedItemsCount = metadataList.size();
        switch (rFetchType) {
            case NONE:
                output = Output.builder().build();
                runContext.metric(Counter.of("fetch.size", 0, "fetch", "false", "store", "false"));
                break;
            case FETCH:
                output = Output.builder()
                    .metadataItems(metadataList)
                    .statusItems(statusList)
                    .size(fetchedItemsCount)
                    .build();
                runContext.metric(Counter.of("fetch.size", fetchedItemsCount, "fetch", "true", "store", "false"));
                break;
            case FETCH_ONE:
                output = Output.builder()
                    .metadataItem(metadataList.isEmpty() ? null : metadataList.getFirst())
                    .statusItem(statusList.isEmpty() ? null : statusList.getFirst())
                    .size(fetchedItemsCount)
                    .build();
                runContext.metric(Counter.of("fetch.size", fetchedItemsCount, "fetch", "true", "store", "false"));
                break;
            case STORE:
                var result = storeResult(metadataList, statusList, runContext);
                int storedItemsCount = result.getValue().intValue();
                output = Output.builder()
                    .uri(result.getKey())
                    .size(storedItemsCount)
                    .build();
                runContext.metric(Counter.of("fetch.size", storedItemsCount, "fetch", "false", "store", "true"));
                break;
            default:
                throw new IllegalStateException("Unexpected fetchType value: " + fetchType);
        }

        return output;
    }

    @Builder
    public record ResourceInfo(Metadata metadata, ResourceStatus status) {
    }

    @Getter
    @Builder
    public static class Output implements io.kestra.core.models.tasks.Output {

        @Schema(
            title = "Metadata list",
            description = "Only available when `fetchType` is set to `FETCH`."
        )
        private final List<Metadata> metadataItems;

        @Schema(
            title = "Single metadata",
            description = "Only available when `fetchType` is set to `FETCH_ONE`."
        )
        private final Metadata metadataItem;

        @Schema(
            title = "Status list",
            description = "Only available when `fetchType` is set to `FETCH`."
        )
        private final List<ResourceStatus> statusItems;

        @Schema(
            title = "Single status",
            description = "Only available when `fetchType` is set to `FETCH_ONE`."
        )
        private final ResourceStatus statusItem;

        @Schema(
            title = "Stored result URI",
            description = "Only available when `fetchType` is set to `STORE`."
        )
        private final URI uri;

        @Schema(
            title = "Resource count"
        )
        private Integer size;
    }

    private Map.Entry<URI, Long> storeResult(List<Metadata> metadataList, List<ResourceStatus> statusList, RunContext runContext) throws IOException {
        var tempFile = runContext.workingDir().createTempFile(".ion").toFile();

        // Combine metadata and status into ResourceInfo objects
        List<ResourceInfo> resourceInfoList = new ArrayList<>();
        for (int i = 0; i < metadataList.size(); i++) {
            resourceInfoList.add(
                ResourceInfo.builder()
                    .metadata(metadataList.get(i))
                    .status(i < statusList.size() ? statusList.get(i) : null)
                    .build()
            );
        }

        try (
            var output = new BufferedWriter(new FileWriter(tempFile), FileSerde.BUFFER_SIZE)
        ) {
            var flowable = Flux
                .create(
                    s ->
                    {
                        resourceInfoList.forEach(s::next);
                        s.complete();
                    },
                    FluxSink.OverflowStrategy.BUFFER
                );

            var count = FileSerde.writeAll(output, flowable);
            var lineCount = count.block();

            output.flush();

            return new AbstractMap.SimpleEntry<>(
                runContext.storage().putFile(tempFile),
                lineCount
            );
        }
    }
}