components/global/core/io.cellery.observability.k8s.client/src/main/java/io/cellery/observability/k8s/client/GetComponentPodsStreamProcessor.java
private void addComponentPods(ComplexEventChunk<StreamEvent> outputStreamEventChunk,
                                  StreamEvent incomingStreamEvent, String componentNameLabel) throws ParseException {
        // Calling the K8s API Servers to fetch component pods
        PodList componentPodList = null;
        try {
            componentPodList = k8sClient.pods()
                    .inNamespace(Constants.NAMESPACE)
                    .withLabel(Constants.CELLERY_OBSERVABILITY_INSTANCE_LABEL)
                    .withLabel(componentNameLabel)
                    .withField(Constants.STATUS_FIELD, Constants.STATUS_FIELD_RUNNING_VALUE)
                    .list();
        } catch (Throwable e) {
            logger.error("Failed to fetch current pods for components", e);
        }

        if (componentPodList != null) {
            for (Pod pod : componentPodList.getItems()) {
                String kind = pod.getMetadata().getLabels()
                        .getOrDefault(Constants.CELLERY_OBSERVABILITY_INSTANCE_KIND_LABEL, "");
                if (logger.isDebugEnabled()) {
                    logger.debug("Added event - pod " + pod.getMetadata().getName() + " belonging to " + kind + " " +
                            pod.getMetadata().getLabels()
                                    .getOrDefault(Constants.CELLERY_OBSERVABILITY_INSTANCE_LABEL, "") +
                            " of type " +
                            (Constants.CELLERY_OBSERVABILITY_COMPONENT_NAME_LABEL.equals(componentNameLabel)
                                    ? "component" : "gateway") + " to the event");
                }

                Object[] newData = new Object[6];
                newData[0] = pod.getMetadata().getLabels().getOrDefault(
                        Constants.CELLERY_OBSERVABILITY_INSTANCE_LABEL, "");
                newData[1] = Utils.getComponentName(pod);
                newData[2] = pod.getMetadata().getName();
                newData[3] = kind;
                newData[4] = new SimpleDateFormat(Constants.K8S_DATE_FORMAT, Locale.US)
                        .parse(pod.getMetadata().getCreationTimestamp()).getTime();
                newData[5] = pod.getSpec().getNodeName();

                StreamEvent streamEventCopy = streamEventCloner.copyStreamEvent(incomingStreamEvent);
                complexEventPopulater.populateComplexEvent(streamEventCopy, newData);
                outputStreamEventChunk.add(streamEventCopy);
            }
        }
    }
/**
     * Add component pods to output stream event chunk.
     * A new event will be cloned from the incoming stream event for each pod and added to the output event chunk.
     *
     * @param outputStreamEventChunk The output stream event chunk which will be sent to the next processor
     * @param incomingStreamEvent    The incoming stream event which will be cloned and used
     * @param componentNameLabel     The name of the label applied to store the component/gateway name
     */
Add component pods to output stream event chunk.
A new event will be cloned from the incoming stream event for each pod and added to the output event chunk.

@param outputStreamEventChunk The output stream event chunk which will be sent to the next processor
@param incomingStreamEvent    The incoming stream event which will be cloned and used
@param componentNameLabel     The name of the label applied to store the component/gateway name
Add component pods to output stream event chunk . A new event will be cloned from the incoming stream event for each pod and added to the output event chunk . @param outputStreamEventChunk The output stream event chunk which will be sent to the next processor @param incomingStreamEvent The incoming stream event which will be cloned and used @param componentNameLabel The name of the label applied to store the component / gateway name
private void addComponentPods(ComplexEventChunk<StreamEvent> outputStreamEventChunk,
                                  StreamEvent incomingStreamEvent, String componentNameLabel) throws ParseException {
        PodList componentPodList = null;
        try {
            componentPodList = k8sClient.pods()
                    .inNamespace(Constants.NAMESPACE)
                    .withLabel(Constants.CELLERY_OBSERVABILITY_INSTANCE_LABEL)
                    .withLabel(componentNameLabel)
                    .withField(Constants.STATUS_FIELD, Constants.STATUS_FIELD_RUNNING_VALUE)
                    .list();
        } catch (Throwable e) {
            logger.error("Failed to fetch current pods for components", e);
        }
        if (componentPodList != null) {
            for (Pod pod : componentPodList.getItems()) {
                String kind = pod.getMetadata().getLabels()
                        .getOrDefault(Constants.CELLERY_OBSERVABILITY_INSTANCE_KIND_LABEL, "");
                if (logger.isDebugEnabled()) {
                    logger.debug("Added event - pod " + pod.getMetadata().getName() + " belonging to " + kind + " " +
                            pod.getMetadata().getLabels()
                                    .getOrDefault(Constants.CELLERY_OBSERVABILITY_INSTANCE_LABEL, "") +
                            " of type " +
                            (Constants.CELLERY_OBSERVABILITY_COMPONENT_NAME_LABEL.equals(componentNameLabel)
                                    ? "component" : "gateway") + " to the event");
                }
                Object[] newData = new Object[6];
                newData[0] = pod.getMetadata().getLabels().getOrDefault(
                        Constants.CELLERY_OBSERVABILITY_INSTANCE_LABEL, "");
                newData[1] = Utils.getComponentName(pod);
                newData[2] = pod.getMetadata().getName();
                newData[3] = kind;
                newData[4] = new SimpleDateFormat(Constants.K8S_DATE_FORMAT, Locale.US)
                        .parse(pod.getMetadata().getCreationTimestamp()).getTime();
                newData[5] = pod.getSpec().getNodeName();
                StreamEvent streamEventCopy = streamEventCloner.copyStreamEvent(incomingStreamEvent);
                complexEventPopulater.populateComplexEvent(streamEventCopy, newData);
                outputStreamEventChunk.add(streamEventCopy);
            }
        }
    }
private void addComponentPods ( ComplexEventChunk < StreamEvent > outputStreamEventChunk , StreamEvent incomingStreamEvent , String componentNameLabel ) throws ParseException { PodList componentPodList = null ; try { componentPodList = k8sClient . pods ( ) . inNamespace ( Constants . NAMESPACE ) . withLabel ( Constants . CELLERY_OBSERVABILITY_INSTANCE_LABEL ) . withLabel ( componentNameLabel ) . withField ( Constants . STATUS_FIELD , Constants . STATUS_FIELD_RUNNING_VALUE ) . list ( ) ; } catch ( Throwable e ) { logger . error ( "Failed to fetch current pods for components" , e ) ; } if ( componentPodList != null ) { for ( Pod pod : componentPodList . getItems ( ) ) { String kind = pod . getMetadata ( ) . getLabels ( ) . getOrDefault ( Constants . CELLERY_OBSERVABILITY_INSTANCE_KIND_LABEL , "" ) ; if ( logger . isDebugEnabled ( ) ) { logger . debug ( "Added event - pod " + pod . getMetadata ( ) . getName ( ) + " belonging to " + kind + " " + pod . getMetadata ( ) . getLabels ( ) . getOrDefault ( Constants . CELLERY_OBSERVABILITY_INSTANCE_LABEL , "" ) + " of type " + ( Constants . CELLERY_OBSERVABILITY_COMPONENT_NAME_LABEL . equals ( componentNameLabel ) ? "component" : "gateway" ) + " to the event" ) ; } Object [ ] newData = new Object [ 6 ] ; newData [ 0 ] = pod . getMetadata ( ) . getLabels ( ) . getOrDefault ( Constants . CELLERY_OBSERVABILITY_INSTANCE_LABEL , "" ) ; newData [ 1 ] = Utils . getComponentName ( pod ) ; newData [ 2 ] = pod . getMetadata ( ) . getName ( ) ; newData [ 3 ] = kind ; newData [ 4 ] = new SimpleDateFormat ( Constants . K8S_DATE_FORMAT , Locale . US ) . parse ( pod . getMetadata ( ) . getCreationTimestamp ( ) ) . getTime ( ) ; newData [ 5 ] = pod . getSpec ( ) . getNodeName ( ) ; StreamEvent streamEventCopy = streamEventCloner . copyStreamEvent ( incomingStreamEvent ) ; complexEventPopulater . populateComplexEvent ( streamEventCopy , newData ) ; outputStreamEventChunk . add ( streamEventCopy ) ; } } }
Add component pods to output stream event chunk.
Add component pods to output stream event chunk .
// Calling the K8s API Servers to fetch component pods