backend/src/main/java/dev/dinauer/service/PodService.java

package dev.dinauer.service;

import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.function.Consumer;

import jakarta.enterprise.context.ApplicationScoped;
import jakarta.inject.Inject;

import io.fabric8.kubernetes.api.model.Pod;
import io.fabric8.kubernetes.api.model.apps.Deployment;
import io.fabric8.kubernetes.api.model.apps.StatefulSet;
import io.fabric8.kubernetes.client.Watch;
import io.fabric8.kubernetes.client.Watcher;
import io.fabric8.kubernetes.client.dsl.AppsAPIGroupDSL;
import io.fabric8.kubernetes.client.dsl.LogWatch;
import io.fabric8.kubernetes.client.dsl.PodResource;
import io.smallrye.mutiny.Uni;

import dev.dinauer.inspect.log.model.KubernetesLog;
import dev.dinauer.inspect.utils.InputStreamWatcher;
import dev.dinauer.utils.ClientProvider;

@ApplicationScoped
public class PodService implements ResourceService<Pod>
{
    @Inject
    ClientProvider clientProvider;

    @Inject
    LogParser logParser;

    @Inject
    dev.dinauer.PodResource podResource;

    @Override
    public void delete(String name, String namespace)
    {
        Optional<Pod> podOptional = findOptionalByNameAndNamespace(name, namespace);
        if (podOptional.isPresent())
        {
            Pod pod = podOptional.get();
            clientProvider.getClient().pods().inNamespace(pod.getMetadata().getNamespace()).withName(pod.getMetadata().getName()).delete();
        }
    }

    @Override
    public Pod findByNameAndNamespace(String name, String namespace)
    {
        return clientProvider.getClient().pods().inNamespace(namespace).withName(name).get();
    }

    public List<Pod> findByNamespace(String namespace)
    {
        if (namespace.equals("_all"))
        {
            return clientProvider.getClient().pods().inAnyNamespace().list().getItems();
        }
        return clientProvider.getClient().pods().inNamespace(namespace).list().getItems();
    }

    public List<Pod> findByStatefulSet(String name, String namespace)
    {
        try (AppsAPIGroupDSL apps = clientProvider.getClient().apps())
        {
            StatefulSet set = apps.statefulSets().inNamespace(namespace).withName(name).get();
            if (set != null)
            {
                return findByNamespaceAndLabels(namespace, set.getSpec().getSelector().getMatchLabels());
            }
            return null;
        }
    }

    public List<Pod> findByDeployment(String namespace, String name)
    {
        try (AppsAPIGroupDSL apps = clientProvider.getClient().apps())
        {
            Deployment set = apps.deployments().inNamespace(namespace).withName(name).get();
            if (set != null)
            {
                return findByNamespaceAndLabels(namespace, set.getSpec().getSelector().getMatchLabels());
            }
            return null;
        }
    }

    public List<Pod> findByNamespaceAndLabels(String namespace, Map<String, String> labels)
    {
        return clientProvider.getClient().pods().inNamespace(namespace).withLabels(labels).list().getItems();
    }

    public List<Pod> findByLabels(Map<String, String> labels)
    {
        return clientProvider.getClient().pods().inAnyNamespace().withLabels(labels).list().getItems();
    }

    public List<Pod> findAll()
    {
        return clientProvider.getClient().pods().inAnyNamespace().list().getItems();
    }

    @Override
    public Optional<Pod> findOptionalByNameAndNamespace(String name, String namespace)
    {
        return Optional.ofNullable(findByNameAndNamespace(name, namespace));
    }

    @Override
    public Watch watch(String namespace, Watcher<Pod> watcher)
    {
        String version = clientProvider.getClient().pods().inAnyNamespace().list().getMetadata().getResourceVersion();
        if (namespace.equals("_all"))
        {
            return clientProvider.getClient().pods().inAnyNamespace().withResourceVersion(version).watch(watcher);
        }
        return clientProvider.getClient().pods().inNamespace(namespace).withResourceVersion(version).watch(watcher);
    }

    public Optional<Pod> findPodById(String id)
    {
        for (Pod pod : clientProvider.getClient().pods().inAnyNamespace().list().getItems())
        {
            if (pod.getMetadata().getUid().equals(id))
            {
                return Optional.of(pod);
            }
        }
        return Optional.empty();
    }

    public List<KubernetesLog> getLogs(PodResource resource)
    {
        List<String> existingLogs = Arrays.stream(resource.usingTimestamps().tailingLines(200).getLog().split("\n")).toList();
        return logParser.toLog(existingLogs, resource.get().getMetadata().getName());
    }

    public LogWatch watchLogs(PodResource resource, Consumer<List<KubernetesLog>> consume)
    {
        LogWatch watch = resource.usingTimestamps().tailingLines(0).watchLog();
        Uni.createFrom()
            .voidItem()
            .invoke(() -> {
                InputStreamWatcher.watch(watch.getOutput(), line -> consume.accept(logParser.toLog(line, resource.get().getMetadata().getName())));
            })
            .subscribe()
            .with(result -> {}, error -> {});
        return watch;
    }
}