Skip to content

chore: remove analytics execution from the critical path #39757

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
vsvamsi1 merged 3 commits into from
Mar 18, 2025
Merged

chore: remove analytics execution from the critical path #39757

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
9e7d287
added further telemetry
vsvamsi1 Mar 17, 2025
3c32cb1
running analystics on a seperate thread without blocking the return o…
vsvamsi1 Mar 17, 2025
ddf93dd
removed scheduler for db call
vsvamsi1 Mar 18, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions ...smith-interfaces/src/main/java/com/appsmith/external/constants/spans/ce/ActionSpanCE.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,12 @@ public class ActionSpanCE {
public static final String GET_ENVIRONMENT_ID = APPSMITH_SPAN_PREFIX + "getEnvironmentId";
public static final String POPULATED_EXECUTE_ACTION_DTO_MONO =
APPSMITH_SPAN_PREFIX + "populatedExecuteActionDTOMono";

public static final String VALIDATE_AUTHENTICATION_DATASOURCE_STORAGE =
APPSMITH_SPAN_PREFIX + "validateAuthenticationDatasourceStorage";
public static final String VERIFY_DATASOURCE_AND_MAKE_REQUEST =
APPSMITH_SPAN_PREFIX + "verifyDatasourceAndMakeRequest";
public static final String SEND_EXECUTE_ANALYTICS_EVENT = APPSMITH_SPAN_PREFIX + "sendExecuteAnalyticsEvent";
public static final String POPULATE_AND_EXECUTE_ACTION = APPSMITH_SPAN_PREFIX + "populateAndExecuteAction";
public static final String GET_VALID_ACTION_FOR_EXECUTION = APPSMITH_SPAN_PREFIX + "getValidActionForExecution";
public static final String GET_CACHED_PLUGIN_FOR_ACTION_EXECUTION =
Expand Down
25 changes: 14 additions & 11 deletions ...-server/src/main/java/com/appsmith/server/services/ce/DatasourceContextServiceCEImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -557,17 +557,20 @@ public Mono<DatasourceContext<?>> getDatasourceContext(DatasourceStorage datasou
public Mono<DatasourceContext<?>> getRemoteDatasourceContext(Plugin plugin, DatasourceStorage datasourceStorage) {
final DatasourceContext<ExecutePluginDTO> datasourceContext = new DatasourceContext<>();

return configService.getInstanceId().map(instanceId -> {
ExecutePluginDTO executePluginDTO = new ExecutePluginDTO();
executePluginDTO.setInstallationKey(instanceId);
executePluginDTO.setPluginName(plugin.getPluginName());
executePluginDTO.setPluginVersion(plugin.getVersion());
executePluginDTO.setDatasource(new RemoteDatasourceDTO(
datasourceStorage.getDatasourceId(), datasourceStorage.getDatasourceConfiguration()));
datasourceContext.setConnection(executePluginDTO);

return datasourceContext;
});
return configService
.getInstanceId()
.subscribeOn(Schedulers.boundedElastic())

@ApekshaBhosale ApekshaBhosale Mar 18, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@vsvamsi1 why did you introduced new thread here?

@vsvamsi1 vsvamsi1 Mar 18, 2025

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Because this is a configService.getInstanceId() a db call, Schedulers.boundedElastic() threads are designed blocking/longer execution tasks.

@ApekshaBhosale ApekshaBhosale Mar 18, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@NilanshBansal then why here? here as well only datasourceContext getting returned

@NilanshBansal NilanshBansal Mar 18, 2025 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@vsvamsi1 we can avoid subscribing this on a scheduler as the ReactiveCrudRepository is already non blocking and handling the database calls. Also, the instanceId is already cached and it is only the first time that a database call is made
cc: @ApekshaBhosale

@vsvamsi1 vsvamsi1 Mar 18, 2025

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@NilanshBansal Cool i will remove the scheduler part of it but can you point me to where the instanceId is cached?

.map(instanceId -> {
ExecutePluginDTO executePluginDTO = new ExecutePluginDTO();
executePluginDTO.setInstallationKey(instanceId);
executePluginDTO.setPluginName(plugin.getPluginName());
executePluginDTO.setPluginVersion(plugin.getVersion());
executePluginDTO.setDatasource(new RemoteDatasourceDTO(
datasourceStorage.getDatasourceId(), datasourceStorage.getDatasourceConfiguration()));
datasourceContext.setConnection(executePluginDTO);

return datasourceContext;
});
}

/**
Expand Down
46 changes: 29 additions & 17 deletions ...-server/src/main/java/com/appsmith/server/solutions/ce/ActionExecutionSolutionCEImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,8 @@
import reactor.core.observability.micrometer.Micrometer;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import reactor.util.function.Tuple2;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
Expand Down Expand Up @@ -360,6 +362,7 @@ public Mono<ActionExecutionResult> executeAction(
: getCachedPluginForActionExecution(datasourceStorageMono)
.name(GET_CACHED_PLUGIN_FOR_ACTION_EXECUTION)
.tap(Micrometer.observation(observationRegistry));

Mono<PluginExecutor> pluginExecutorMono = pluginExecutorHelper
.getPluginExecutor(pluginMono)
.name(GET_PLUGIN_EXECUTOR)
Expand All @@ -375,7 +378,6 @@ public Mono<ActionExecutionResult> executeAction(
executeActionMetaDTO.getHeaders())
.name(GET_ACTION_EXECUTION_RESULT)
.tap(Micrometer.observation(observationRegistry));

Mono<Map> editorConfigLabelMapMono = getEditorConfigLabelMap(datasourceStorageMono);

return actionExecutionResultMono
Expand Down Expand Up @@ -764,6 +766,8 @@ protected Mono<ActionExecutionResult> verifyDatasourceAndMakeRequest(

Mono<ActionExecutionResult> executionMono = authenticationValidator
.validateAuthentication(datasourceStorage)
.name(VALIDATE_AUTHENTICATION_DATASOURCE_STORAGE)
.tap(Micrometer.observation(observationRegistry))
.zipWhen(validatedDatasource -> datasourceContextService
.getDatasourceContext(validatedDatasource, plugin)
.tag("plugin", plugin.getPackageName())
Expand Down Expand Up @@ -909,37 +913,45 @@ protected Mono<ActionExecutionResult> getActionExecutionResult(
Mono<ActionDTO> actionDTOWithAutoGeneratedHeadersMono =
setAutoGeneratedHeaders(plugin, actionDTO, httpHeaders);

Mono<ActionExecutionResult> actionExecutionResultMono =
actionDTOWithAutoGeneratedHeadersMono.flatMap(actionDTO1 -> verifyDatasourceAndMakeRequest(
Mono<ActionExecutionResult> actionExecutionResultMono = actionDTOWithAutoGeneratedHeadersMono
.flatMap(actionDTO1 -> verifyDatasourceAndMakeRequest(
executeActionDTO, actionDTO, datasourceStorage, plugin, pluginExecutor)
.timeout(Duration.ofMillis(timeoutDuration)));
.timeout(Duration.ofMillis(timeoutDuration)))
.name(VERIFY_DATASOURCE_AND_MAKE_REQUEST)
.tap(Micrometer.observation(observationRegistry));

ActionConfiguration finalRawActionConfiguration = rawActionConfiguration;
return actionExecutionResultMono
.onErrorMap(executionExceptionMapper(actionDTO, timeoutDuration))
.onErrorResume(executionExceptionHandler(actionDTO))
.elapsed()
// Now send the analytics event for this execution
.flatMap(tuple1 -> {
.map(tuple1 -> {
Long timeElapsed = tuple1.getT1();
ActionExecutionResult result = tuple1.getT2();

log.debug(
"{}: Action {} with id {} execution time : {} ms",
Thread.currentThread().getName(),
actionDTO.getName(),
actionDTO.getId(),
timeElapsed);

return sendExecuteAnalyticsEvent(
return tuple1;
})
.doOnSuccess(tuple2 -> {
Long timeElapsed = tuple2.getT1();
ActionExecutionResult result = tuple2.getT2();
// Runs the analytics in the separate thread and immediately return the execution result
sendExecuteAnalyticsEvent(
actionDTO,
datasourceStorage,
executeActionDTO,
result,
timeElapsed,
finalRawActionConfiguration)
.thenReturn(result);
});
.name(SEND_EXECUTE_ANALYTICS_EVENT)
.tap(Micrometer.observation(observationRegistry))
.subscribeOn(Schedulers.boundedElastic())
.subscribe();
})
.map(Tuple2::getT2);
});
}

Expand Down Expand Up @@ -1097,16 +1109,16 @@ private Mono<ActionExecutionRequest> sendExecuteAnalyticsEvent(
request.setProperties(stringProperties);
}

return Mono.justOrEmpty(actionDTO.getApplicationId())
Mono<Application> applicationMono = Mono.justOrEmpty(actionDTO.getApplicationId())
.flatMap(applicationService::findById)
Comment thread
vsvamsi1 marked this conversation as resolved.
.defaultIfEmpty(new Application())
.flatMap(application -> Mono.zip(
Mono.just(application),
.defaultIfEmpty(new Application());
return Mono.zip(
applicationMono,
sessionUserService.getCurrentUser(),
newPageService.getNameByPageId(actionDTO.getPageId(), executeActionDto.getViewMode()),
pluginService.getByIdWithoutPermissionCheck(actionDTO.getPluginId()),
datasourceStorageService.getEnvironmentNameFromEnvironmentIdForAnalytics(
datasourceStorage.getEnvironmentId())))
datasourceStorage.getEnvironmentId()))
.flatMap(tuple -> {
final Application application = tuple.getT1();
final User user = tuple.getT2();
Expand Down