feat: added onboarding setup for Producer for Messaging queues (#6236)

* fix: added onboarding setup for producer/consumer for Messaging queues

* fix: polled onboarding status api

* feat: added onboarding status api with useQueury functions and updated endpoints

* feat: added onboarding status api util for attribute data

* feat: refactoring and url query changes

* feat: changed start and end time to nanosecond for api payload

* feat: added comment description

frontend/src/api/messagingQueues/onboarding/getOnboardingStatus.ts

@@ -0,0 +1,37 @@
+import { ApiBaseInstance } from 'api';
+import { ErrorResponseHandler } from 'api/ErrorResponseHandler';
+import { AxiosError } from 'axios';
+import { SOMETHING_WENT_WRONG } from 'constants/api';
+import { ErrorResponse, SuccessResponse } from 'types/api';
+
+export interface OnboardingStatusResponse {
+	status: string;
+	data: {
+		attribute?: string;
+		error_message?: string;
+		status?: string;
+	}[];
+}
+
+const getOnboardingStatus = async (props: {
+	start: number;
+	end: number;
+}): Promise<SuccessResponse<OnboardingStatusResponse> | ErrorResponse> => {
+	try {
+		const response = await ApiBaseInstance.post(
+			'/messaging-queues/kafka/onboarding/consumers',
+			props,
+		);
+
+		return {
+			statusCode: 200,
+			error: null,
+			message: response.data.status,
+			payload: response.data,
+		};
+	} catch (error) {
+		return ErrorResponseHandler((error as AxiosError) || SOMETHING_WENT_WRONG);
+	}
+};
+
+export default getOnboardingStatus;

frontend/src/constants/query.ts

@@ -38,4 +38,6 @@ export enum QueryParams {
 	selectedTimelineQuery = 'selectedTimelineQuery',
 	ruleType = 'ruleType',
 	configDetail = 'configDetail',
+	getStartedSource = 'getStartedSource',
+	getStartedSourceService = 'getStartedSourceService',
 }

frontend/src/container/OnboardingContainer/Modules/APM/Java/md-docs/SpringBoot/Kubernetes/springBoot-kubernetes-runApplication-producer.md

@@ -0,0 +1,29 @@
+ 
+
+Once you are done intrumenting your Java application, you can run it using the below commands
+
+**Note:**
+- Ensure you have Java and Maven installed. Compile your Java producer applications: Ensure your producer and consumer apps are compiled and ready to run.
+
+**Run Producer App with Java Agent:**
+
+```bash
+java -javaagent:/path/to/opentelemetry-javaagent.jar \
+     -Dotel.service.name=producer-svc \
+     -Dotel.traces.exporter=otlp \
+     -Dotel.metrics.exporter=otlp \
+     -Dotel.logs.exporter=otlp \
+     -jar /path/to/your/producer.jar
+```
+
+<path> - update it to the path where you downloaded the Java JAR agent in previous step
+<my-app> - Jar file of your application
+
+&nbsp;
+
+**Note:**
+- In case you're dockerising your application, make sure to dockerise it along with OpenTelemetry instrumentation done in previous step.
+
+&nbsp;
+
+If you encounter any difficulties, please consult the [troubleshooting section](https://signoz.io/docs/instrumentation/springboot/#troubleshooting-your-installation) for assistance.

frontend/src/container/OnboardingContainer/Steps/ConnectionStatus/ConnectionStatus.tsx

@@ -6,11 +6,15 @@ import {
 	LoadingOutlined,
 } from '@ant-design/icons';
 import logEvent from 'api/common/logEvent';
+import { QueryParams } from 'constants/query';
 import Header from 'container/OnboardingContainer/common/Header/Header';
 import { useOnboardingContext } from 'container/OnboardingContainer/context/OnboardingContext';
+import { useOnboardingStatus } from 'hooks/messagingQueue / onboarding/useOnboardingStatus';
 import { useQueryService } from 'hooks/useQueryService';
 import useResourceAttribute from 'hooks/useResourceAttribute';
 import { convertRawQueriesToTraceSelectedTags } from 'hooks/useResourceAttribute/utils';
+import useUrlQuery from 'hooks/useUrlQuery';
+import { getAttributeDataFromOnboardingStatus } from 'pages/MessagingQueues/MessagingQueuesUtils';
 import { useEffect, useMemo, useState } from 'react';
 import { useDispatch, useSelector } from 'react-redux';
 import { AppState } from 'store/reducers';
@@ -27,6 +31,9 @@ export default function ConnectionStatus(): JSX.Element {
 		GlobalReducer
 	>((state) => state.globalTime);
 
+	const urlQuery = useUrlQuery();
+	const getStartedSource = urlQuery.get(QueryParams.getStartedSource);
+
 	const {
 		serviceName,
 		selectedDataSource,
@@ -57,8 +64,65 @@ export default function ConnectionStatus(): JSX.Element {
 		maxTime,
 		selectedTime,
 		selectedTags,
+		options: {
+			enabled: getStartedSource !== 'kafka',
+		},
 	});
 
+	const [pollInterval, setPollInterval] = useState<number | false>(10000);
+	const {
+		data: onbData,
+		error: onbErr,
+		isFetching: onbFetching,
+	} = useOnboardingStatus({
+		enabled: getStartedSource === 'kafka',
+		refetchInterval: pollInterval,
+	});
+
+	const [
+		shouldRetryOnboardingCall,
+		setShouldRetryOnboardingCall,
+	] = useState<boolean>(false);
+
+	useEffect(() => {
+		// runs only when the caller is coming from 'kafka' i.e. coming from Messaging Queues - setup helper
+		if (getStartedSource === 'kafka') {
+			if (onbData?.statusCode !== 200) {
+				setShouldRetryOnboardingCall(true);
+			} else if (onbData?.payload?.status === 'success') {
+				const attributeData = getAttributeDataFromOnboardingStatus(
+					onbData?.payload,
+				);
+				if (attributeData.overallStatus === 'success') {
+					setLoading(false);
+					setIsReceivingData(true);
+					setPollInterval(false);
+				} else {
+					setShouldRetryOnboardingCall(true);
+				}
+			}
+		}
+	}, [
+		shouldRetryOnboardingCall,
+		onbData,
+		onbErr,
+		onbFetching,
+		getStartedSource,
+	]);
+
+	useEffect(() => {
+		if (retryCount < 0 && getStartedSource === 'kafka') {
+			setPollInterval(false);
+			setLoading(false);
+		}
+	}, [retryCount, getStartedSource]);
+
+	useEffect(() => {
+		if (getStartedSource === 'kafka' && !onbFetching) {
+			setRetryCount((prevCount) => prevCount - 1);
+		}
+	}, [getStartedSource, onbData, onbFetching]);
+
 	const renderDocsReference = (): JSX.Element => {
 		switch (selectedDataSource?.name) {
 			case 'java':
@@ -192,25 +256,27 @@ export default function ConnectionStatus(): JSX.Element {
 	useEffect(() => {
 		let pollingTimer: string | number | NodeJS.Timer | undefined;
 
-		if (loading) {
-			pollingTimer = setInterval(() => {
-				// Trigger a refetch with the updated parameters
-				const updatedMinTime = (Date.now() - 15 * 60 * 1000) * 1000000;
-				const updatedMaxTime = Date.now() * 1000000;
+		if (getStartedSource !== 'kafka') {
+			if (loading) {
+				pollingTimer = setInterval(() => {
+					// Trigger a refetch with the updated parameters
+					const updatedMinTime = (Date.now() - 15 * 60 * 1000) * 1000000;
+					const updatedMaxTime = Date.now() * 1000000;
 
-				const payload = {
-					maxTime: updatedMaxTime,
-					minTime: updatedMinTime,
-					selectedTime,
-				};
+					const payload = {
+						maxTime: updatedMaxTime,
+						minTime: updatedMinTime,
+						selectedTime,
+					};
 
-				dispatch({
-					type: UPDATE_TIME_INTERVAL,
-					payload,
-				});
-			}, pollingInterval); // Same interval as pollingInterval
-		} else if (!loading && pollingTimer) {
-			clearInterval(pollingTimer);
+					dispatch({
+						type: UPDATE_TIME_INTERVAL,
+						payload,
+					});
+				}, pollingInterval); // Same interval as pollingInterval
+			} else if (!loading && pollingTimer) {
+				clearInterval(pollingTimer);
+			}
 		}
 
 		// Clean up the interval when the component unmounts
@@ -221,15 +287,24 @@ export default function ConnectionStatus(): JSX.Element {
 	}, [refetch, selectedTags, selectedTime, loading]);
 
 	useEffect(() => {
-		verifyApplicationData(data);
+		if (getStartedSource !== 'kafka') {
+			verifyApplicationData(data);
+		}
 		// eslint-disable-next-line react-hooks/exhaustive-deps
 	}, [isServiceLoading, data, error, isError]);
 
 	useEffect(() => {
-		refetch();
+		if (getStartedSource !== 'kafka') {
+			refetch();
+		}
 		// eslint-disable-next-line react-hooks/exhaustive-deps
 	}, []);
 
+	const isQueryServiceLoading = useMemo(
+		() => isServiceLoading || loading || onbFetching,
+		[isServiceLoading, loading, onbFetching],
+	);
+
 	return (
 		<div className="connection-status-container">
 			<div className="full-docs-link">{renderDocsReference()}</div>
@@ -250,14 +325,14 @@ export default function ConnectionStatus(): JSX.Element {
 					<div className="label"> Status </div>
 
 					<div className="status">
-						{(loading || isServiceLoading) && <LoadingOutlined />}
-						{!(loading || isServiceLoading) && isReceivingData && (
+						{isQueryServiceLoading && <LoadingOutlined />}
+						{!isQueryServiceLoading && isReceivingData && (
 							<>
 								<CheckCircleTwoTone twoToneColor="#52c41a" />
 								<span> Success </span>
 							</>
 						)}
-						{!(loading || isServiceLoading) && !isReceivingData && (
+						{!isQueryServiceLoading && !isReceivingData && (
 							<>
 								<CloseCircleTwoTone twoToneColor="#e84749" />
 								<span> Failed </span>
@@ -269,11 +344,11 @@ export default function ConnectionStatus(): JSX.Element {
 					<div className="label"> Details </div>
 
 					<div className="details">
-						{(loading || isServiceLoading) && <div> Waiting for Update </div>}
-						{!(loading || isServiceLoading) && isReceivingData && (
+						{isQueryServiceLoading && <div> Waiting for Update </div>}
+						{!isQueryServiceLoading && isReceivingData && (
 							<div> Received data from the application successfully. </div>
 						)}
-						{!(loading || isServiceLoading) && !isReceivingData && (
+						{!isQueryServiceLoading && !isReceivingData && (
 							<div> Could not detect the install </div>
 						)}
 					</div>

frontend/src/container/OnboardingContainer/Steps/DataSource/DataSource.styles.scss

@@ -74,4 +74,11 @@ div[class*='-setup-instructions-container'] {
 	.dataSourceName {
 		color: var(--bg-slate-500);
 	}
-}
+}
+
+.supported-languages-container {
+	.disabled {
+		cursor: not-allowed;
+		opacity: 0.5;
+	}
+}

frontend/src/container/OnboardingContainer/Steps/DataSource/DataSource.tsx

@@ -6,6 +6,7 @@ import { LoadingOutlined } from '@ant-design/icons';
 import { Button, Card, Form, Input, Select, Space, Typography } from 'antd';
 import logEvent from 'api/common/logEvent';
 import cx from 'classnames';
+import { QueryParams } from 'constants/query';
 import ROUTES from 'constants/routes';
 import { useOnboardingContext } from 'container/OnboardingContainer/context/OnboardingContext';
 import { useCases } from 'container/OnboardingContainer/OnboardingContainer';
@@ -13,8 +14,10 @@ import {
 	getDataSources,
 	getSupportedFrameworks,
 	hasFrameworks,
+	messagingQueueKakfaSupportedDataSources,
 } from 'container/OnboardingContainer/utils/dataSourceUtils';
 import { useNotifications } from 'hooks/useNotifications';
+import useUrlQuery from 'hooks/useUrlQuery';
 import { Blocks, Check } from 'lucide-react';
 import { useEffect, useState } from 'react';
 import { useTranslation } from 'react-i18next';
@@ -33,6 +36,8 @@ export default function DataSource(): JSX.Element {
 	const { t } = useTranslation(['common']);
 	const history = useHistory();
 
+	const getStartedSource = useUrlQuery().get(QueryParams.getStartedSource);
+
 	const {
 		serviceName,
 		selectedModule,
@@ -150,13 +155,19 @@ export default function DataSource(): JSX.Element {
 						className={cx(
 							'supported-language',
 							selectedDataSource?.name === dataSource.name ? 'selected' : '',
+							getStartedSource === 'kafka' &&
+								!messagingQueueKakfaSupportedDataSources.includes(dataSource?.id || '')
+								? 'disabled'
+								: '',
 						)}
 						key={dataSource.name}
 						onClick={(): void => {
-							updateSelectedFramework(null);
-							updateSelectedEnvironment(null);
-							updateSelectedDataSource(dataSource);
-							form.setFieldsValue({ selectFramework: null });
+							if (getStartedSource !== 'kafka') {
+								updateSelectedFramework(null);
+								updateSelectedEnvironment(null);
+								updateSelectedDataSource(dataSource);
+								form.setFieldsValue({ selectFramework: null });
+							}
 						}}
 					>
 						<div>

frontend/src/container/OnboardingContainer/Steps/MarkdownStep/MarkdownStep.tsx

@@ -1,5 +1,6 @@
 /* eslint-disable @typescript-eslint/ban-ts-comment */
 import { MarkdownRenderer } from 'components/MarkdownRenderer/MarkdownRenderer';
+import { QueryParams } from 'constants/query';
 import { ApmDocFilePaths } from 'container/OnboardingContainer/constants/apmDocFilePaths';
 import { AwsMonitoringDocFilePaths } from 'container/OnboardingContainer/constants/awsMonitoringDocFilePaths';
 import { AzureMonitoringDocFilePaths } from 'container/OnboardingContainer/constants/azureMonitoringDocFilePaths';
@@ -10,6 +11,7 @@ import {
 	useOnboardingContext,
 } from 'container/OnboardingContainer/context/OnboardingContext';
 import { ModulesMap } from 'container/OnboardingContainer/OnboardingContainer';
+import useUrlQuery from 'hooks/useUrlQuery';
 import { useEffect, useState } from 'react';
 
 export interface IngestionInfoProps {
@@ -31,6 +33,12 @@ export default function MarkdownStep(): JSX.Element {
 
 	const [markdownContent, setMarkdownContent] = useState('');
 
+	const urlQuery = useUrlQuery();
+	const getStartedSource = urlQuery.get(QueryParams.getStartedSource);
+	const getStartedSourceService = urlQuery.get(
+		QueryParams.getStartedSourceService,
+	);
+
 	const { step } = activeStep;
 
 	const getFilePath = (): any => {
@@ -54,6 +62,12 @@ export default function MarkdownStep(): JSX.Element {
 
 		path += `_${step?.id}`;
 
+		if (
+			getStartedSource === 'kafka' &&
+			path === 'APM_java_springBoot_kubernetes_recommendedSteps_runApplication' // todo: Sagar - Make this generic logic in followup PRs
+		) {
+			path += `_${getStartedSourceService}`;
+		}
 		return path;
 	};
 

frontend/src/container/OnboardingContainer/constants/apmDocFilePaths.ts

@@ -252,6 +252,7 @@ import APM_java_springBoot_docker_recommendedSteps_runApplication from '../Modul
 import APM_java_springBoot_kubernetes_recommendedSteps_setupOtelCollector from '../Modules/APM/Java/md-docs/SpringBoot/Kubernetes/springBoot-kubernetes-installOtelCollector.md';
 import APM_java_springBoot_kubernetes_recommendedSteps_instrumentApplication from '../Modules/APM/Java/md-docs/SpringBoot/Kubernetes/springBoot-kubernetes-instrumentApplication.md';
 import APM_java_springBoot_kubernetes_recommendedSteps_runApplication from '../Modules/APM/Java/md-docs/SpringBoot/Kubernetes/springBoot-kubernetes-runApplication.md';
+import APM_java_springBoot_kubernetes_recommendedSteps_runApplication_producer from '../Modules/APM/Java/md-docs/SpringBoot/Kubernetes/springBoot-kubernetes-runApplication-producer.md';
 // SpringBoot-LinuxAMD64-quickstart
 import APM_java_springBoot_linuxAMD64_quickStart_instrumentApplication from '../Modules/APM/Java/md-docs/SpringBoot/LinuxAMD64/QuickStart/springBoot-linuxamd64-quickStart-instrumentApplication.md';
 import APM_java_springBoot_linuxAMD64_quickStart_runApplication from '../Modules/APM/Java/md-docs/SpringBoot/LinuxAMD64/QuickStart/springBoot-linuxamd64-quickStart-runApplication.md';
@@ -1053,6 +1054,7 @@ export const ApmDocFilePaths = {
 	APM_java_springBoot_kubernetes_recommendedSteps_setupOtelCollector,
 	APM_java_springBoot_kubernetes_recommendedSteps_instrumentApplication,
 	APM_java_springBoot_kubernetes_recommendedSteps_runApplication,
+	APM_java_springBoot_kubernetes_recommendedSteps_runApplication_producer,
 
 	// SpringBoot-LinuxAMD64-recommended
 	APM_java_springBoot_linuxAMD64_recommendedSteps_setupOtelCollector,

frontend/src/container/OnboardingContainer/utils/dataSourceUtils.ts

@@ -399,3 +399,5 @@ export const moduleRouteMap = {
 	[ModulesMap.AwsMonitoring]: ROUTES.GET_STARTED_AWS_MONITORING,
 	[ModulesMap.AzureMonitoring]: ROUTES.GET_STARTED_AZURE_MONITORING,
 };
+
+export const messagingQueueKakfaSupportedDataSources = ['java'];

frontend/src/hooks/messagingQueue / onboarding/useOnboardingStatus.tsx

@@ -0,0 +1,22 @@
+import getOnboardingStatus, {
+	OnboardingStatusResponse,
+} from 'api/messagingQueues/onboarding/getOnboardingStatus';
+import { useQuery, UseQueryOptions, UseQueryResult } from 'react-query';
+import { ErrorResponse, SuccessResponse } from 'types/api';
+
+type UseOnboardingStatus = (
+	options?: UseQueryOptions<
+		SuccessResponse<OnboardingStatusResponse> | ErrorResponse
+	>,
+) => UseQueryResult<SuccessResponse<OnboardingStatusResponse> | ErrorResponse>;
+
+export const useOnboardingStatus: UseOnboardingStatus = (options) =>
+	useQuery<SuccessResponse<OnboardingStatusResponse> | ErrorResponse>({
+		queryKey: ['onboardingStatus'],
+		queryFn: () =>
+			getOnboardingStatus({
+				start: (Date.now() - 15 * 60 * 1000) * 1_000_000,
+				end: Date.now() * 1_000_000,
+			}),
+		...options,
+	});

frontend/src/pages/MessagingQueues/MessagingQueues.tsx

@@ -4,6 +4,7 @@ import { ExclamationCircleFilled } from '@ant-design/icons';
 import { Color } from '@signozhq/design-tokens';
 import { Button, Modal } from 'antd';
 import logEvent from 'api/common/logEvent';
+import { QueryParams } from 'constants/query';
 import ROUTES from 'constants/routes';
 import DateTimeSelectionV2 from 'container/TopNav/DateTimeSelectionV2';
 import { Calendar, ListMinus } from 'lucide-react';
@@ -86,7 +87,7 @@ function MessagingQueues(): JSX.Element {
 								type="default"
 								onClick={(): void =>
 									getStartedRedirect(
-										ROUTES.GET_STARTED_APPLICATION_MONITORING,
+										`${ROUTES.GET_STARTED_APPLICATION_MONITORING}?${QueryParams.getStartedSource}=kafka&${QueryParams.getStartedSourceService}=consumer`,
 										'Configure Consumer',
 									)
 								}
@@ -105,7 +106,7 @@ function MessagingQueues(): JSX.Element {
 								type="default"
 								onClick={(): void =>
 									getStartedRedirect(
-										ROUTES.GET_STARTED_APPLICATION_MONITORING,
+										`${ROUTES.GET_STARTED_APPLICATION_MONITORING}?${QueryParams.getStartedSource}=kafka&${QueryParams.getStartedSourceService}=producer`,
 										'Configure Producer',
 									)
 								}
@@ -124,7 +125,7 @@ function MessagingQueues(): JSX.Element {
 								type="default"
 								onClick={(): void =>
 									getStartedRedirect(
-										ROUTES.GET_STARTED_INFRASTRUCTURE_MONITORING,
+										`${ROUTES.GET_STARTED_INFRASTRUCTURE_MONITORING}?${QueryParams.getStartedSource}=kafka&${QueryParams.getStartedSourceService}=kafka`,
 										'Monitor kafka',
 									)
 								}

frontend/src/pages/MessagingQueues/MessagingQueuesUtils.ts

@@ -1,3 +1,4 @@
+import { OnboardingStatusResponse } from 'api/messagingQueues/onboarding/getOnboardingStatus';
 import { QueryParams } from 'constants/query';
 import { PANEL_TYPES } from 'constants/queryBuilder';
 import { GetWidgetQueryBuilderProps } from 'container/MetricsApplication/types';
@@ -344,3 +345,39 @@ export const getMetaDataAndAPIPerView = (
 		},
 	};
 };
+
+interface OnboardingStatusAttributeData {
+	overallStatus: string;
+	allAvailableAttributes: string[];
+	attributeDataWithError: { attributeName: string; errorMsg: string }[];
+}
+
+export const getAttributeDataFromOnboardingStatus = (
+	onboardingStatus?: OnboardingStatusResponse | null,
+): OnboardingStatusAttributeData => {
+	const allAvailableAttributes: string[] = [];
+	const attributeDataWithError: {
+		attributeName: string;
+		errorMsg: string;
+	}[] = [];
+
+	if (onboardingStatus?.data && !isEmpty(onboardingStatus?.data)) {
+		onboardingStatus.data.forEach((status) => {
+			if (status.attribute) {
+				allAvailableAttributes.push(status.attribute);
+				if (status.status === '0') {
+					attributeDataWithError.push({
+						attributeName: status.attribute,
+						errorMsg: status.error_message || '',
+					});
+				}
+			}
+		});
+	}
+
+	return {
+		overallStatus: attributeDataWithError.length ? 'error' : 'success',
+		allAvailableAttributes,
+		attributeDataWithError,
+	};
+};