seecoder-paas/src/main/java/cn/seecoder/paas/service/impl/MetricsServiceImpl.java

package cn.seecoder.paas.service.impl;

import cn.seecoder.paas.data.dao.EnvironmentDAO;
import cn.seecoder.paas.data.dao.MetricsDAO;
import cn.seecoder.paas.data.entity.Environment;
import cn.seecoder.paas.data.entity.Metrics;
import cn.seecoder.paas.service.MetricsService;
import cn.seecoder.paas.service.facade.k8s.PodApi;
import cn.seecoder.paas.service.facade.k8s.model.K8sObjectRequest;
import cn.seecoder.paas.service.facade.k8s.model.Pod;
import cn.seecoder.paas.service.model.converter.MetricsConverter;
import cn.seecoder.paas.service.model.vo.MetricsVO;
import cn.seecoder.paas.util.ApplicationProperties;
import cn.seecoder.paas.util.DateUtil;
import cn.seecoder.paas.util.enums.MetricsSource;
import cn.seecoder.paas.util.enums.MetricsType;
import cn.seecoder.paas.util.enums.ResourceLabel;
import io.fabric8.kubernetes.api.model.Quantity;
import io.fabric8.kubernetes.api.model.metrics.v1beta1.NodeMetrics;
import io.fabric8.kubernetes.api.model.metrics.v1beta1.NodeMetricsList;
import io.fabric8.kubernetes.api.model.metrics.v1beta1.PodMetrics;
import io.fabric8.kubernetes.client.DefaultKubernetesClient;
import org.springframework.beans.factory.annotation.Autowired;

import java.time.LocalDateTime;
import java.util.*;
import java.util.stream.Collectors;

@org.springframework.stereotype.Service
public class MetricsServiceImpl implements MetricsService {

    private final DefaultKubernetesClient fabricK8sClient;

    private final ApplicationProperties applicationProperties;

    private final EnvironmentDAO environmentDAO;

    private final MetricsDAO metricsDAO;

    private final PodApi podApi;

    @Autowired
    public MetricsServiceImpl(DefaultKubernetesClient fabricK8sClient, ApplicationProperties applicationProperties, EnvironmentDAO environmentDAO, PodApi podApi, MetricsDAO metricsDAO) {
        this.fabricK8sClient = fabricK8sClient;
        this.applicationProperties = applicationProperties;
        this.environmentDAO = environmentDAO;
        this.podApi = podApi;
        this.metricsDAO = metricsDAO;
    }

    @Override
    public List<MetricsVO> getMetricsBySourceAndTypeBetween(MetricsSource metricsSource, String metricsSourceValue, MetricsType metricsType, LocalDateTime startTime, LocalDateTime endTime) {
        List<Metrics> metrics = metricsDAO.findByMetricsSourceAndAndMetricsSourceValueAndMetricsTypeAndMetricsTimeBetween(metricsSource, metricsSourceValue, metricsType, startTime, endTime);
        return metrics.stream().map(MetricsConverter::convertToVO).collect(Collectors.toList());
    }

    @Override
    public List<MetricsVO> getMetricsSourceCurrentStatus(MetricsSource metricsSource, String metricsSourceValue) {
        List<MetricsVO> result = new ArrayList<>();
        String deploymentNamespace = applicationProperties.getDeploymentNamespace();
        switch (metricsSource) {
            case POD:
                PodMetrics podMetrics = fabricK8sClient.top().pods().metrics(deploymentNamespace, metricsSourceValue);
                return statPodMetrics(Collections.singletonList(podMetrics), metricsSource, metricsSourceValue);
            case NODE:
                NodeMetrics nodeMetrics = fabricK8sClient.top().nodes().metrics(metricsSourceValue);
                return statNodeMetrics(Collections.singletonList(nodeMetrics), metricsSource, metricsSourceValue);
            case APPLICATION:
                List<Environment> environments = environmentDAO.findAllByAppId(Integer.parseInt(metricsSourceValue));
                List<PodMetrics> podMetricsList = new ArrayList<>();
                environments.forEach(environment -> {
                    String labelKey = ResourceLabel.RESOURCE.getCode();
                    String labelValue = ResourceLabel.RESOURCE.getGenerator().gen("environment", metricsSourceValue, String.valueOf(environment.getId()));
                    K8sObjectRequest request = K8sObjectRequest.builder().namespace(applicationProperties.getDeploymentNamespace()).labels(Collections.singletonMap(labelKey, labelValue)).build();
                    List<Pod> pods = podApi.getByCondition(request);
                    pods.forEach(pod -> {
                        PodMetrics metrics = fabricK8sClient.top().pods().metrics(deploymentNamespace, pod.getName());
                        podMetricsList.add(metrics);
                    });
                });
                return statPodMetrics(podMetricsList, metricsSource, metricsSourceValue);
            case ENVIRONMENT:
                Environment environment = environmentDAO.findById(Integer.parseInt(metricsSourceValue)).orElse(null);
                if (environment == null) {
                    return Collections.emptyList();
                }
                List<PodMetrics> environmentMetricsList = new ArrayList<>();
                String labelKey = ResourceLabel.RESOURCE.getCode();
                String labelValue = ResourceLabel.RESOURCE.getGenerator().gen("environment", String.valueOf(environment.getAppId()), String.valueOf(environment.getId()));
                K8sObjectRequest request = K8sObjectRequest.builder().namespace(applicationProperties.getDeploymentNamespace()).labels(Collections.singletonMap(labelKey, labelValue)).build();
                List<Pod> pods = podApi.getByCondition(request);
                pods.forEach(pod -> {
                    PodMetrics metrics = fabricK8sClient.top().pods().metrics(deploymentNamespace, pod.getName());
                    environmentMetricsList.add(metrics);
                });
                return statPodMetrics(environmentMetricsList, metricsSource, metricsSourceValue);
            case CLUSTER:
                NodeMetricsList nodeMetricsList = fabricK8sClient.top().nodes().metrics();
                return statNodeMetrics(nodeMetricsList.getItems(), metricsSource, metricsSourceValue);
        }
        return result;
    }

    public List<MetricsVO> statPodMetrics(List<PodMetrics> podMetricsList, MetricsSource metricsSource, String metricsSourceValue) {
        if (podMetricsList == null || podMetricsList.isEmpty()) {
            return Collections.emptyList();
        }
        List<MetricsVO> result;
        Map<String, Long> values = new HashMap<>();
        podMetricsList.forEach(podMetrics ->  {
            podMetrics.getContainers().forEach(containerMetrics -> {
                Map<String, Quantity> metricValues = containerMetrics.getUsage();
                metricValues.forEach((key, quantity) -> {
                    if (values.get(key) != null) {
                        values.put(key, values.get(key) + Long.parseLong(quantity.getAmount()));
                    } else {
                        values.put(key, Long.parseLong(quantity.getAmount()));
                    }
                });
            });
        });
        result = values.entrySet().stream().map(entry -> {
            MetricsVO vo = new MetricsVO();
            vo.setMetricsSource(metricsSource);
            vo.setMetricsSourceValue(metricsSourceValue);
            vo.setMetricsType(entry.getKey());
            vo.setMetricsTypeValue(Long.toString(entry.getValue()));
            vo.setMetricsTime(podMetricsList.size() == 1 ? DateUtil.fromUTCTimeString(podMetricsList.get(0).getTimestamp()) : LocalDateTime.now());
            return vo;

        }).collect(Collectors.toList());
        if (podMetricsList.size() == 1) {
            // duration
            MetricsVO durationVO = new MetricsVO();
            durationVO.setMetricsSource(metricsSource);
            durationVO.setMetricsSourceValue(metricsSourceValue);
            durationVO.setMetricsType(MetricsType.DURATION.getCode());
            durationVO.setMetricsTypeValue(podMetricsList.get(0).getWindow().getDuration().toString());
            durationVO.setMetricsTime(DateUtil.fromUTCTimeString(podMetricsList.get(0).getTimestamp()));
            result.add(durationVO);
        }
        return result;
    }

    public List<MetricsVO> statNodeMetrics(List<NodeMetrics> nodeMetricsList, MetricsSource metricsSource, String metricsSourceValue) {
        if (nodeMetricsList == null || nodeMetricsList.isEmpty()) {
            return Collections.emptyList();
        }
        List<MetricsVO> result;
        Map<String, Long> values = new HashMap<>();
        nodeMetricsList.forEach(nodeMetrics ->  {
            Map<String, Quantity> metricValues = nodeMetrics.getUsage();
            metricValues.forEach((key, quantity) -> {
                if (values.get(key) != null) {
                    values.put(key, values.get(key) + Long.parseLong(quantity.getAmount()));
                } else {
                    values.put(key, Long.parseLong(quantity.getAmount()));
                }
            });
        });
        result = values.entrySet().stream().map(entry -> {
            MetricsVO vo = new MetricsVO();
            vo.setMetricsSource(metricsSource);
            vo.setMetricsSourceValue(metricsSourceValue);
            vo.setMetricsType(entry.getKey());
            vo.setMetricsTypeValue(Long.toString(entry.getValue()));
            vo.setMetricsTime(nodeMetricsList.size() == 1 ? DateUtil.fromUTCTimeString(nodeMetricsList.get(0).getTimestamp()) : LocalDateTime.now());
            return vo;

        }).collect(Collectors.toList());
        if (nodeMetricsList.size() == 1) {
            // duration
            MetricsVO durationVO = new MetricsVO();
            durationVO.setMetricsSource(metricsSource);
            durationVO.setMetricsSourceValue(metricsSourceValue);
            durationVO.setMetricsType(MetricsType.DURATION.getCode());
            durationVO.setMetricsTypeValue(nodeMetricsList.get(0).getWindow().getDuration().toString());
            durationVO.setMetricsTime(DateUtil.fromUTCTimeString(nodeMetricsList.get(0).getTimestamp()));
            result.add(durationVO);
        }
        return result;
    }
}