Skip to content

Commit

Permalink
concord-server: close MultipartInput explicitly (#1084)
Browse files Browse the repository at this point in the history 
Remove temporary files as soon as possible. Do not rely on finalizers.
  • Loading branch information
@ibodrov
ibodrov authored Feb 18, 2025
1 parent 34cbc1d commit e33c459
Show file tree
Hide file tree
Showing 11 changed files with 178 additions and 131 deletions.
12 changes: 8 additions & 4 deletions server/impl/src/main/java/com/walmartlabs/concord/server/console/CustomFormService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -93,10 +93,14 @@ public Response continueSession(@Context UriInfo uriInfo,
@PathParam("formName") String formName,
MultipartInput data) {

if (isV2(processInstanceId)) {
return customFormServiceV2.continueSession(uriInfo, headers, processInstanceId, formName, data);
} else {
return customFormServiceV1.continueSession(uriInfo, headers, processInstanceId, formName, data);
try {
if (isV2(processInstanceId)) {
return customFormServiceV2.continueSession(uriInfo, headers, processInstanceId, formName, data);
} else {
return customFormServiceV1.continueSession(uriInfo, headers, processInstanceId, formName, data);
}
} finally {
data.close();
}
}

Expand Down
10 changes: 7 additions & 3 deletions server/impl/src/main/java/com/walmartlabs/concord/server/console/CustomFormServiceV1.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -150,7 +150,7 @@ private FormSessionResponse startSession(ProcessKey processKey, String formName)
// copy shared resources (if present)
copySharedResources(processKey, dst);
} catch (IOException e) {
log.warn("startSession ['{}', '{}'] -> error while preparing a custom form: {}", processKey, formName, e);
log.warn("startSession ['{}', '{}'] -> error while preparing a custom form: {}", processKey, formName, e.getMessage());
throw new ConcordApplicationException("Error while preparing a custom form", e);
}

Expand Down Expand Up @@ -182,8 +182,12 @@ public Response continueSession(@Context UriInfo uriInfo,
@PathParam("formName") String formName,
MultipartInput data) {

ProcessKey processKey = assertKey(processInstanceId);
return continueSession(uriInfo, headers, processKey, formName, MultipartUtils.toMap(data));
try {
ProcessKey processKey = assertKey(processInstanceId);
return continueSession(uriInfo, headers, processKey, formName, MultipartUtils.toMap(data));
} finally {
data.close();
}
}

private Response continueSession(UriInfo uriInfo, HttpHeaders headers, ProcessKey processKey,
Expand Down
8 changes: 6 additions & 2 deletions server/impl/src/main/java/com/walmartlabs/concord/server/console/CustomFormServiceV2.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -173,8 +173,12 @@ public Response continueSession(@Context UriInfo uriInfo,
@PathParam("formName") String formName,
MultipartInput data) {

ProcessKey processKey = assertKey(processInstanceId);
return continueSession(uriInfo, headers, processKey, formName, MultipartUtils.toMap(data));
try {
ProcessKey processKey = assertKey(processInstanceId);
return continueSession(uriInfo, headers, processKey, formName, MultipartUtils.toMap(data));
} finally {
data.close();
}
}

private Response continueSession(UriInfo uriInfo, HttpHeaders headers, ProcessKey processKey,
Expand Down
34 changes: 19 additions & 15 deletions server/impl/src/main/java/com/walmartlabs/concord/server/console/ProcessCardResource.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -135,21 +135,25 @@ public GenericOperationResult access(
public ProcessCardOperationResponse createOrUpdate(
@Parameter(schema = @Schema(type = "object", implementation = ProcessCardRequest.class)) MultipartInput input) throws IOException {

ProcessCardRequest r = ProcessCardRequest.from(input);

UUID orgId = organizationManager.assertAccess(r.getOrgId(), r.getOrgName(), false).getId();
UUID projectId = assertProject(orgId, r);
UUID repoId = getRepo(projectId, r);
String name = r.getName();
Optional<String> entryPoint = Optional.ofNullable(r.getEntryPoint());
String description = r.getDescription();
Map<String, Object> data = r.getData();
UUID id = r.getId();
Integer orderId = r.getOrderId();

try (InputStream icon = r.getIcon();
InputStream form = r.getForm()) {
return processCardManager.createOrUpdate(id, projectId, repoId, name, entryPoint, description, icon, form, data, orderId);
try {
ProcessCardRequest r = ProcessCardRequest.from(input);

UUID orgId = organizationManager.assertAccess(r.getOrgId(), r.getOrgName(), false).getId();
UUID projectId = assertProject(orgId, r);
UUID repoId = getRepo(projectId, r);
String name = r.getName();
Optional<String> entryPoint = Optional.ofNullable(r.getEntryPoint());
String description = r.getDescription();
Map<String, Object> data = r.getData();
UUID id = r.getId();
Integer orderId = r.getOrderId();

try (InputStream icon = r.getIcon();
InputStream form = r.getForm()) {
return processCardManager.createOrUpdate(id, projectId, repoId, name, entryPoint, description, icon, form, data, orderId);
}
} finally {
input.close();
}
}

Expand Down
1 change: 1 addition & 0 deletions server/impl/src/main/java/com/walmartlabs/concord/server/org/secret/GetDataRequest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
import org.jboss.resteasy.plugins.providers.multipart.MultipartInput;

public class GetDataRequest {

public static GetDataRequest from(MultipartInput input) {
return new GetDataRequest(input);
}
Expand Down
52 changes: 29 additions & 23 deletions server/impl/src/main/java/com/walmartlabs/concord/server/org/secret/SecretResource.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -135,6 +135,8 @@ public SecretOperationResponse create(@PathParam("orgName") @ConcordKey String o
}
} catch (IOException e) {
throw new ConcordApplicationException("Error while processing the request: " + e.getMessage(), e);
} finally {
input.close();
}
}

Expand Down Expand Up @@ -201,33 +203,37 @@ public Response getData(@PathParam("orgName") @ConcordKey String orgName,
@PathParam("secretName") @ConcordKey String secretName,
@Parameter(schema = @Schema(type = "object", implementation = GetDataRequest.class)) MultipartInput input) {

GetDataRequest request = GetDataRequest.from(input);
try {
GetDataRequest request = GetDataRequest.from(input);

OrganizationEntry org = orgManager.assertAccess(orgName, false);
String password = request.getPassword();
OrganizationEntry org = orgManager.assertAccess(orgName, false);
String password = request.getPassword();

SecretDao.SecretDataEntry entry;
try {
entry = secretManager.getRaw(SecretManager.AccessScope.apiRequest(), org.getId(), secretName, password);
if (entry == null) {
throw new WebApplicationException("Secret not found: " + secretName, Status.NOT_FOUND);
SecretDao.SecretDataEntry entry;
try {
entry = secretManager.getRaw(SecretManager.AccessScope.apiRequest(), org.getId(), secretName, password);
if (entry == null) {
throw new WebApplicationException("Secret not found: " + secretName, Status.NOT_FOUND);
}
} catch (SecurityException e) {
log.warn("fetchSecret -> error: {}", e.getMessage());
throw new SecretException("Error while fetching a secret '" + secretName + "': " + e.getMessage());
} catch (ValidationErrorsException e) {
log.warn("fetchSecret -> error: {}", e.getMessage());
return null;
}
} catch (SecurityException e) {
log.warn("fetchSecret -> error: {}", e.getMessage());
throw new SecretException("Error while fetching a secret '" + secretName + "': " + e.getMessage());
} catch (ValidationErrorsException e) {
log.warn("fetchSecret -> error: {}", e.getMessage());
return null;
}

try {
return Response.ok((StreamingOutput) output -> output.write(entry.getData()),
MediaType.APPLICATION_OCTET_STREAM)
.header(Constants.Headers.SECRET_TYPE, entry.getType().name())
.build();
} catch (Exception e) {
log.error("fetchSecret ['{}'] -> error while fetching a secret", secretName, e);
throw new ConcordApplicationException("Error while fetching a secret '" + secretName + "': " + e.getMessage());
try {
return Response.ok((StreamingOutput) output -> output.write(entry.getData()),
MediaType.APPLICATION_OCTET_STREAM)
.header(Constants.Headers.SECRET_TYPE, entry.getType().name())
.build();
} catch (Exception e) {
log.error("fetchSecret ['{}'] -> error while fetching a secret", secretName, e);
throw new ConcordApplicationException("Error while fetching a secret '" + secretName + "': " + e.getMessage());
}
} finally {
input.close();
}
}

Expand Down
56 changes: 30 additions & 26 deletions server/impl/src/main/java/com/walmartlabs/concord/server/org/secret/SecretResourceV2.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -104,34 +104,38 @@ public GenericOperationResult update(@PathParam("orgName") @ConcordKey String or
@PathParam("secretName") @ConcordKey String secretName,
@Parameter(schema = @Schema(type = "object", implementation = Object.class)) MultipartInput input) {

OrganizationEntry org = orgManager.assertAccess(orgName, true);
Set<UUID> projectIds = getProjectIds(
org.getId(),
MultipartUtils.getUUIDList(input, Constants.Multipart.PROJECT_IDS),
MultipartUtils.getStringList(input, Constants.Multipart.PROJECT_NAMES),
MultipartUtils.getUuid(input, Constants.Multipart.PROJECT_ID),
MultipartUtils.getString(input, Constants.Multipart.PROJECT_NAME)
);
try {
SecretUpdateParams newSecretParams = SecretUpdateParams.builder()
.newOrgId(MultipartUtils.getUuid(input, Constants.Multipart.ORG_ID))
.newOrgName(MultipartUtils.getString(input, Constants.Multipart.ORG_NAME))
.newProjectIds(projectIds)
.removeProjectLink(MultipartUtils.getBoolean(input, "removeProjectLink", false))
.newOwnerId(MultipartUtils.getUuid(input, "ownerId"))
.currentPassword(MultipartUtils.getString(input, Constants.Multipart.STORE_PASSWORD))
.newPassword(MultipartUtils.getString(input, Constants.Multipart.NEW_STORE_PASSWORD))
.newSecret(buildSecret(input))
.newName(MultipartUtils.getString(input, Constants.Multipart.NAME))
.newVisibility(SecretResourceUtils.getVisibility(input))
.build();

secretManager.update(org.getId(), secretName, newSecretParams);
} catch (IOException e) {
throw new ConcordApplicationException("Error while processing the request: " + e.getMessage(), e);
}
OrganizationEntry org = orgManager.assertAccess(orgName, true);
Set<UUID> projectIds = getProjectIds(
org.getId(),
MultipartUtils.getUUIDList(input, Constants.Multipart.PROJECT_IDS),
MultipartUtils.getStringList(input, Constants.Multipart.PROJECT_NAMES),
MultipartUtils.getUuid(input, Constants.Multipart.PROJECT_ID),
MultipartUtils.getString(input, Constants.Multipart.PROJECT_NAME)
);
try {
SecretUpdateParams newSecretParams = SecretUpdateParams.builder()
.newOrgId(MultipartUtils.getUuid(input, Constants.Multipart.ORG_ID))
.newOrgName(MultipartUtils.getString(input, Constants.Multipart.ORG_NAME))
.newProjectIds(projectIds)
.removeProjectLink(MultipartUtils.getBoolean(input, "removeProjectLink", false))
.newOwnerId(MultipartUtils.getUuid(input, "ownerId"))
.currentPassword(MultipartUtils.getString(input, Constants.Multipart.STORE_PASSWORD))
.newPassword(MultipartUtils.getString(input, Constants.Multipart.NEW_STORE_PASSWORD))
.newSecret(buildSecret(input))
.newName(MultipartUtils.getString(input, Constants.Multipart.NAME))
.newVisibility(SecretResourceUtils.getVisibility(input))
.build();

secretManager.update(org.getId(), secretName, newSecretParams);
} catch (IOException e) {
throw new ConcordApplicationException("Error while processing the request: " + e.getMessage(), e);
}

return new GenericOperationResult(OperationResult.UPDATED);
return new GenericOperationResult(OperationResult.UPDATED);
} finally {
input.close();
}
}

public Secret buildSecret(MultipartInput input) throws IOException {
Expand Down
74 changes: 41 additions & 33 deletions server/impl/src/main/java/com/walmartlabs/concord/server/process/ProcessResource.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -271,26 +271,30 @@ public StartProcessResponse start(@Parameter(schema = @Schema(type = "object", i
@Parameter(hidden = true) @Deprecated @QueryParam("out") String[] out,
@Context HttpServletRequest request) {

boolean sync2 = MultipartUtils.getBoolean(input, Constants.Multipart.SYNC, false);
if (sync || sync2) {
throw syncIsForbidden();
}

Payload payload;
try {
payload = payloadManager.createPayload(input, request);
boolean sync2 = MultipartUtils.getBoolean(input, Constants.Multipart.SYNC, false);
if (sync || sync2) {
throw syncIsForbidden();
}

// TODO remove after deprecating the old endpoints
payload = PayloadBuilder.basedOn(payload)
.parentInstanceId(parentInstanceId)
.mergeOutExpressions(out)
.build();
} catch (IOException e) {
log.error("start -> error creating a payload: {}", e.getMessage());
throw new ConcordApplicationException("Error creating a payload", e);
}
Payload payload;
try {
payload = payloadManager.createPayload(input, request);

return toResponse(processManager.start(payload));
// TODO remove after deprecating the old endpoints
payload = PayloadBuilder.basedOn(payload)
.parentInstanceId(parentInstanceId)
.mergeOutExpressions(out)
.build();
} catch (IOException e) {
log.error("start -> error creating a payload: {}", e.getMessage());
throw new ConcordApplicationException("Error creating a payload", e);
}

return toResponse(processManager.start(payload));
} finally {
input.close();
}
}

/**
Expand All @@ -310,27 +314,31 @@ public StartProcessResponse start(@PathParam("entryPoint") String entryPoint,
@Deprecated @DefaultValue("false") @QueryParam("sync") boolean sync,
@QueryParam("out") String[] out) {

if (sync) {
throw syncIsForbidden();
}
try {
if (sync) {
throw syncIsForbidden();
}

assertPartialKey(parentInstanceId);
assertPartialKey(parentInstanceId);

PartialProcessKey processKey = PartialProcessKey.from(UUID.randomUUID());
PartialProcessKey processKey = PartialProcessKey.from(UUID.randomUUID());

UUID orgId = OrganizationManager.DEFAULT_ORG_ID;
EntryPoint ep = payloadManager.parseEntryPoint(processKey, orgId, entryPoint);
UserPrincipal userPrincipal = UserPrincipal.assertCurrent();
UUID orgId = OrganizationManager.DEFAULT_ORG_ID;
EntryPoint ep = payloadManager.parseEntryPoint(processKey, orgId, entryPoint);
UserPrincipal userPrincipal = UserPrincipal.assertCurrent();

Payload payload;
try {
payload = payloadManager.createPayload(processKey, parentInstanceId, userPrincipal.getId(), userPrincipal.getUsername(), ep, input, out);
} catch (IOException e) {
log.error("start ['{}'] -> error creating a payload: {}", entryPoint, e);
throw new ConcordApplicationException("Error creating a payload", e);
}
Payload payload;
try {
payload = payloadManager.createPayload(processKey, parentInstanceId, userPrincipal.getId(), userPrincipal.getUsername(), ep, input, out);
} catch (IOException e) {
log.error("start ['{}'] -> error creating a payload: {}", entryPoint, e);
throw new ConcordApplicationException("Error creating a payload", e);
}

return toResponse(processManager.start(payload));
return toResponse(processManager.start(payload));
} finally {
input.close();
}
}

/**
Expand Down
42 changes: 23 additions & 19 deletions ...ain/java/com/walmartlabs/concord/server/process/checkpoint/ProcessCheckpointResource.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -112,25 +112,29 @@ public ResumeProcessResponse restore(@PathParam("id") UUID instanceId,
public void uploadCheckpoint(@PathParam("id") UUID instanceId,
@Parameter(schema = @Schema(type = "object", implementation = Object.class)) MultipartInput input) {

// TODO replace with ProcessKeyCache
ProcessEntry entry = processManager.assertProcess(instanceId);
ProcessKey processKey = new ProcessKey(entry.instanceId(), entry.createdAt());

UUID checkpointId = MultipartUtils.assertUuid(input, "id");
UUID correlationId = MultipartUtils.assertUuid(input, "correlationId");
String checkpointName = MultipartUtils.assertString(input, "name");
try (InputStream data = MultipartUtils.assertStream(input, "data");
TemporaryPath tmpIn = IOUtils.tempFile("checkpoint", ".zip")) {

Files.copy(data, tmpIn.path(), StandardCopyOption.REPLACE_EXISTING);
checkpointManager.importCheckpoint(processKey, checkpointId, correlationId, checkpointName, tmpIn.path());
} catch (ValidationErrorsException e) {
throw new ConcordApplicationException(e.getMessage(), Response.Status.BAD_REQUEST);
} catch (IOException e) {
log.error("uploadCheckpoint ['{}'] -> error", processKey, e);
throw new ConcordApplicationException("upload error: " + e.getMessage());
try {
// TODO replace with ProcessKeyCache
ProcessEntry entry = processManager.assertProcess(instanceId);
ProcessKey processKey = new ProcessKey(entry.instanceId(), entry.createdAt());

UUID checkpointId = MultipartUtils.assertUuid(input, "id");
UUID correlationId = MultipartUtils.assertUuid(input, "correlationId");
String checkpointName = MultipartUtils.assertString(input, "name");
try (InputStream data = MultipartUtils.assertStream(input, "data");
TemporaryPath tmpIn = IOUtils.tempFile("checkpoint", ".zip")) {

Files.copy(data, tmpIn.path(), StandardCopyOption.REPLACE_EXISTING);
checkpointManager.importCheckpoint(processKey, checkpointId, correlationId, checkpointName, tmpIn.path());
} catch (ValidationErrorsException e) {
throw new ConcordApplicationException(e.getMessage(), Response.Status.BAD_REQUEST);
} catch (IOException e) {
log.error("uploadCheckpoint ['{}'] -> error", processKey, e);
throw new ConcordApplicationException("upload error: " + e.getMessage());
}

log.info("uploadCheckpoint ['{}'] -> done", processKey);
} finally {
input.close();
}

log.info("uploadCheckpoint ['{}'] -> done", processKey);
}
}
Loading

0 comments on commit e33c459

Please sign in to comment.