Skip to content
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

Jump to bottom

Support name filter at get workflows #1525

Merged
merged 8 commits into from
Feb 4, 2021
Merged

Support name filter at get workflows #1525

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
8afc292
Add `name` parameter
Jan 21, 2021
e71ba3d
Support name pattern match
Jan 21, 2021
49492fd
Add a new test case
Jan 22, 2021
af05bb1
Refactor
Jan 22, 2021
983b31a
Add a comment
Jan 22, 2021
601788e
Reanme the perameter name
Jan 22, 2021
3ea375a
Refactor
Jan 25, 2021
102cb3a
rename a parameter
Feb 1, 2021
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
42 changes: 37 additions & 5 deletions digdag-core/src/main/java/io/digdag/core/database/DatabaseProjectStoreManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,8 +39,6 @@
import org.skife.jdbi.v2.sqlobject.stringtemplate.UseStringTemplate3StatementLocator;
import org.skife.jdbi.v2.tweak.ResultSetMapper;

import javax.activation.DataSource;

import java.nio.ByteBuffer;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
Expand All @@ -53,8 +51,8 @@
import java.util.Map;
import java.util.stream.Collectors;

import static java.util.Locale.ENGLISH;
import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.Locale.ENGLISH;

public class DatabaseProjectStoreManager
extends BasicDatabaseStoreManager<DatabaseProjectStoreManager.Dao>
Expand Down Expand Up @@ -292,10 +290,17 @@ public StoredWorkflowDefinitionWithProject getLatestWorkflowDefinitionByName(int
public List<StoredWorkflowDefinitionWithProject> getLatestActiveWorkflowDefinitions(
int pageSize,
Optional<Long> lastId,
Optional<String> namePattern,
AccessController.ListFilter acFilter)
throws ResourceNotFoundException
{
return autoCommit((handle, dao) -> dao.getLatestActiveWorkflowDefinitions(siteId, pageSize, lastId.or(0L), acFilter.getSql()));
return autoCommit((handle, dao) -> dao.getLatestActiveWorkflowDefinitions(
siteId,
pageSize,
lastId.or(0L),
generatePartialMatchPattern(namePattern),
acFilter.getSql())
);
}

@DigdagTimed(value = "dpst_", category = "db", appendMethodName = true)
Expand Down Expand Up @@ -355,6 +360,19 @@ public TimeZoneMap getWorkflowTimeZonesByIdList(List<Long> defIdList)
Map<Long, ZoneId> map = IdTimeZone.listToMap(list);
return new TimeZoneMap(map);
}

private String generatePartialMatchPattern(Optional<String> pattern)
Copy link
Contributor

Choose a reason for hiding this comment

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

👍

{
// If provided pattern is absent or empty string, just set '%'
// so that the pattern does not affect to a where clause.
return !pattern.or("").isEmpty() ? "%" + escapeLikePattern(pattern.get()) + "%" : "%";
}

private String escapeLikePattern(String pattern)
{
return pattern.replace("%", "\\%")
.replace("_", "\\_");
}
}

private static class IdTimeZone
Expand Down Expand Up @@ -571,13 +589,20 @@ public interface H2Dao
" join projects proj on a.project_id = proj.id" +
" join workflow_configs wc on wc.id = wd.config_id" +
" where wd.id \\> :lastId" +
// `workflow_definitions` table has a composite index
// for `revision_id` and `name` (`workflow_definitions_on_revision_id_and_name`).
// And the index is used for filter by `revision_id` and `name`.
// Since this query always limits the records by `revision_id` (the latest revision's one),
// partial matching of `name` (e.g. '%test%') can be accepted.
" and wd.name like :namePattern" +
" and <acFilter>" +
" order by wd.id" +
" limit :limit")
List<StoredWorkflowDefinitionWithProject> getLatestActiveWorkflowDefinitions(
@Bind("siteId") int siteId,
@Bind("limit") int limit,
@Bind("lastId") long lastId,
@Bind("namePattern") String namePattern,
@Define("acFilter") String acFilter);
}

Expand Down Expand Up @@ -613,6 +638,12 @@ public interface PgDao
" group by r.project_id" +
" )) " +
" and wf.id \\> :lastId" +
// `workflow_definitions` table has a composite index
// for `revision_id` and `name` (`workflow_definitions_on_revision_id_and_name`).
// And the index is used for filter by `revision_id` and `name`.
// Since this query always limits the records by `revision_id` (the latest revision's one),
// partial matching of `name` (e.g. '%test%') can be accepted.
" and wf.name like :namePattern" +
" and <acFilter>" +
" order by wf.id" +
" limit :limit" +
Expand All @@ -625,6 +656,7 @@ List<StoredWorkflowDefinitionWithProject> getLatestActiveWorkflowDefinitions(
@Bind("siteId") int siteId,
@Bind("limit") int limit,
@Bind("lastId") long lastId,
@Bind("namePattern") String namePattern,
@Define("acFilter") String acFilter);
}

Expand Down Expand Up @@ -728,7 +760,7 @@ List<StoredProject> getProjects(
" limit 1")
StoredWorkflowDefinitionWithProject getLatestWorkflowDefinitionByName(@Bind("siteId") int siteId, @Bind("projId") int projId, @Bind("name") String name);

List<StoredWorkflowDefinitionWithProject> getLatestActiveWorkflowDefinitions(int siteId, int limit, long lastId, String acFilter);
List<StoredWorkflowDefinitionWithProject> getLatestActiveWorkflowDefinitions(int siteId, int limit, long lastId, String namePattern, String acFilter);

// getWorkflowDetailsById is same with getWorkflowDetailsByIdInternal
// excepting site_id check
Expand Down
16 changes: 7 additions & 9 deletions digdag-core/src/main/java/io/digdag/core/repository/ProjectStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,12 +1,10 @@
package io.digdag.core.repository;

import java.util.List;
import java.util.Map;
import java.time.ZoneId;

import com.google.common.base.Optional;
import io.digdag.spi.ac.AccessController;

import java.util.List;

public interface ProjectStore
{
List<StoredProject> getProjects(int pageSize, Optional<Integer> lastId, AccessController.ListFilter acFilter);
Expand Down Expand Up @@ -56,16 +54,16 @@ byte[] getRevisionArchiveData(int revId)
List<StoredWorkflowDefinition> getWorkflowDefinitions(int revId, int pageSize, Optional<Long> lastId, AccessController.ListFilter acFilter);

StoredWorkflowDefinition getWorkflowDefinitionByName(int revId, String name)
throws ResourceNotFoundException;
throws ResourceNotFoundException;

StoredWorkflowDefinitionWithProject getWorkflowDefinitionById(long wfId)
throws ResourceNotFoundException;
throws ResourceNotFoundException;

StoredWorkflowDefinitionWithProject getLatestWorkflowDefinitionByName(int projId, String name)
throws ResourceNotFoundException;
throws ResourceNotFoundException;

List<StoredWorkflowDefinitionWithProject> getLatestActiveWorkflowDefinitions(int pageSize, Optional<Long> lastId, AccessController.ListFilter acFilter)
throws ResourceNotFoundException;
List<StoredWorkflowDefinitionWithProject> getLatestActiveWorkflowDefinitions(int pageSize, Optional<Long> lastId, Optional<String> namePattern, AccessController.ListFilter acFilter)
throws ResourceNotFoundException;

TimeZoneMap getWorkflowTimeZonesByIdList(List<Long> defIdList);
}
121 changes: 107 additions & 14 deletions digdag-core/src/test/java/io/digdag/core/database/DatabaseProjectStoreManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,17 +1,45 @@
package io.digdag.core.database;

import java.util.*;
import com.google.common.base.Optional;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
import io.digdag.core.repository.ImmutableProject;
import io.digdag.core.repository.ImmutableRevision;
import io.digdag.core.repository.ImmutableWorkflowDefinition;
import io.digdag.core.repository.Project;
import io.digdag.core.repository.ProjectControl;
import io.digdag.core.repository.ProjectMap;
import io.digdag.core.repository.ProjectStore;
import io.digdag.core.repository.ProjectStoreManager;
import io.digdag.core.repository.Revision;
import io.digdag.core.repository.StoredProject;
import io.digdag.core.repository.StoredRevision;
import io.digdag.core.repository.StoredWorkflowDefinition;
import io.digdag.core.repository.StoredWorkflowDefinitionWithProject;
import io.digdag.core.repository.TimeZoneMap;
import io.digdag.core.repository.WorkflowDefinition;
import io.digdag.core.schedule.SchedulerManager;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.time.Instant;
import java.util.List;
import java.util.concurrent.atomic.AtomicReference;
import org.skife.jdbi.v2.IDBI;
import org.junit.*;
import com.google.common.base.Optional;
import com.google.common.collect.*;
import io.digdag.core.repository.*;
import io.digdag.core.schedule.*;

import static io.digdag.core.database.DatabaseTestingUtils.assertConflict;
import static io.digdag.core.database.DatabaseTestingUtils.assertEmpty;
import static io.digdag.core.database.DatabaseTestingUtils.assertNotConflict;
import static io.digdag.core.database.DatabaseTestingUtils.assertNotFound;
import static io.digdag.core.database.DatabaseTestingUtils.createRevision;
import static io.digdag.core.database.DatabaseTestingUtils.createWorkflow;
import static io.digdag.core.database.DatabaseTestingUtils.setupDatabase;
import static java.nio.charset.StandardCharsets.UTF_8;
import static io.digdag.core.database.DatabaseTestingUtils.*;
import static org.junit.Assert.*;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertTrue;

public class DatabaseProjectStoreManagerTest
{
Expand Down Expand Up @@ -209,10 +237,10 @@ public void testGetAndNotFounds()
assertEquals(ImmutableList.of(wf4), store.getWorkflowDefinitions(rev3.getId(), 100, Optional.of(wf3.getId()), () -> "true"));
assertEmpty(anotherSite.getWorkflowDefinitions(rev3.getId(), 100, Optional.absent(), () -> "true"));

assertEquals(ImmutableList.of(wfDetails1, wfDetails3, wfDetails4), store.getLatestActiveWorkflowDefinitions(100, Optional.absent(), () -> "true"));
assertEquals(ImmutableList.of(wfDetails1), store.getLatestActiveWorkflowDefinitions(1, Optional.absent(), () -> "true"));
assertEquals(ImmutableList.of(wfDetails4), store.getLatestActiveWorkflowDefinitions(100, Optional.of(wfDetails3.getId()), () -> "true"));
assertEmpty(anotherSite.getLatestActiveWorkflowDefinitions(100, Optional.absent(), () -> "true"));
assertEquals(ImmutableList.of(wfDetails1, wfDetails3, wfDetails4), store.getLatestActiveWorkflowDefinitions(100, Optional.absent(), Optional.absent(), () -> "true"));
assertEquals(ImmutableList.of(wfDetails1), store.getLatestActiveWorkflowDefinitions(1, Optional.absent(), Optional.absent(), () -> "true"));
assertEquals(ImmutableList.of(wfDetails4), store.getLatestActiveWorkflowDefinitions(100, Optional.of(wfDetails3.getId()), Optional.absent(), () -> "true"));
assertEmpty(anotherSite.getLatestActiveWorkflowDefinitions(100, Optional.absent(), Optional.absent(), () -> "true"));

////
// public simple getters
Expand Down Expand Up @@ -274,6 +302,71 @@ public void testGetAndNotFounds()
});
}

@Test
public void testGetActiveWorkflows()
throws Exception
{
factory.begin(() -> {
Project srcProj1 = Project.of("proj1");
Revision srcRev1 = createRevision("rev1");
WorkflowDefinition srcWf1 = createWorkflow("wf1");
WorkflowDefinition srcWf2 = createWorkflow("test_wf2");
WorkflowDefinition srcWf3 = createWorkflow("test_wf3");
WorkflowDefinition srcWf4 = createWorkflow("wf%4");
WorkflowDefinition srcWf5 = createWorkflow("wf_5");
WorkflowDefinition srcWf6 = createWorkflow("wf%_6");
final AtomicReference<StoredRevision> revRef = new AtomicReference<>();
final AtomicReference<StoredWorkflowDefinition> wfRef1 = new AtomicReference<>();
final AtomicReference<StoredWorkflowDefinition> wfRef2 = new AtomicReference<>();
final AtomicReference<StoredWorkflowDefinition> wfRef3 = new AtomicReference<>();
final AtomicReference<StoredWorkflowDefinition> wfRef4 = new AtomicReference<>();
final AtomicReference<StoredWorkflowDefinition> wfRef5 = new AtomicReference<>();
final AtomicReference<StoredWorkflowDefinition> wfRef6 = new AtomicReference<>();
StoredProject proj1 = store.putAndLockProject(
srcProj1,
(store, stored) -> {
ProjectControl lock = new ProjectControl(store, stored);
assertNotConflict(() -> {
revRef.set(lock.insertRevision(srcRev1));
wfRef1.set(lock.insertWorkflowDefinitions(revRef.get(), ImmutableList.of(srcWf1), sm, Instant.now()).get(0));
wfRef2.set(lock.insertWorkflowDefinitions(revRef.get(), ImmutableList.of(srcWf2), sm, Instant.now()).get(0));
wfRef3.set(lock.insertWorkflowDefinitions(revRef.get(), ImmutableList.of(srcWf3), sm, Instant.now()).get(0));
wfRef4.set(lock.insertWorkflowDefinitions(revRef.get(), ImmutableList.of(srcWf4), sm, Instant.now()).get(0));
wfRef5.set(lock.insertWorkflowDefinitions(revRef.get(), ImmutableList.of(srcWf5), sm, Instant.now()).get(0));
wfRef6.set(lock.insertWorkflowDefinitions(revRef.get(), ImmutableList.of(srcWf6), sm, Instant.now()).get(0));
});
return lock.get();
});
StoredWorkflowDefinition wf1 = wfRef1.get();
StoredWorkflowDefinition wf2 = wfRef2.get();
StoredWorkflowDefinition wf3 = wfRef3.get();
StoredWorkflowDefinition wf4 = wfRef4.get();
StoredWorkflowDefinition wf5 = wfRef5.get();
StoredWorkflowDefinition wf6 = wfRef6.get();
StoredWorkflowDefinitionWithProject wfDetails1 = StoredWorkflowDefinitionWithProject.of(wf1, proj1, srcRev1);
StoredWorkflowDefinitionWithProject wfDetails2 = StoredWorkflowDefinitionWithProject.of(wf2, proj1, srcRev1);
StoredWorkflowDefinitionWithProject wfDetails3 = StoredWorkflowDefinitionWithProject.of(wf3, proj1, srcRev1);
StoredWorkflowDefinitionWithProject wfDetails4 = StoredWorkflowDefinitionWithProject.of(wf4, proj1, srcRev1);
StoredWorkflowDefinitionWithProject wfDetails5 = StoredWorkflowDefinitionWithProject.of(wf5, proj1, srcRev1);
StoredWorkflowDefinitionWithProject wfDetails6 = StoredWorkflowDefinitionWithProject.of(wf6, proj1, srcRev1);

assertEquals(ImmutableList.of(wfDetails1, wfDetails2, wfDetails3, wfDetails4, wfDetails5, wfDetails6),
store.getLatestActiveWorkflowDefinitions(100, Optional.absent(), Optional.absent(), () -> "true"));
assertEquals(ImmutableList.of(wfDetails1, wfDetails2, wfDetails3, wfDetails4, wfDetails5, wfDetails6),
store.getLatestActiveWorkflowDefinitions(100, Optional.absent(), Optional.fromNullable(""), () -> "true"));
assertEquals(ImmutableList.of(wfDetails2, wfDetails3),
store.getLatestActiveWorkflowDefinitions(100, Optional.absent(), Optional.fromNullable("test"), () -> "true"));
assertEquals(ImmutableList.of(wfDetails4, wfDetails6),
store.getLatestActiveWorkflowDefinitions(100, Optional.absent(), Optional.fromNullable("%"), () -> "true"));
assertEquals(ImmutableList.of(wfDetails2, wfDetails3, wfDetails5, wfDetails6),
store.getLatestActiveWorkflowDefinitions(100, Optional.absent(), Optional.fromNullable("_"), () -> "true"));
assertEquals(ImmutableList.of(wfDetails6),
store.getLatestActiveWorkflowDefinitions(100, Optional.absent(), Optional.fromNullable("%_"), () -> "true"));
assertEquals(ImmutableList.of(),
store.getLatestActiveWorkflowDefinitions(100, Optional.absent(), Optional.fromNullable("*"), () -> "true"));
});
}

@Test
public void testRevisionArchiveData()
throws Exception
Expand Down Expand Up @@ -328,7 +421,7 @@ public void testDeleteProject()

// listing doesn't include deleted projects
assertEquals(ImmutableList.of(), store.getProjects(100, Optional.absent(), () -> "true"));
assertEquals(ImmutableList.of(), store.getLatestActiveWorkflowDefinitions(100, Optional.absent(), () -> "true"));
assertEquals(ImmutableList.of(), store.getLatestActiveWorkflowDefinitions(100, Optional.absent(), Optional.absent(), () -> "true"));

// lookup by project/revision id succeeds and deletedAt is set
StoredProject deletedProj = store.getProjectById(deletingProject.getId());
Expand Down
6 changes: 5 additions & 1 deletion digdag-server/src/main/java/io/digdag/server/rs/WorkflowResource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,7 +113,10 @@ public RestWorkflowDefinitionCollection getWorkflowDefinitions(
@ApiParam(value="list workflows whose id is grater than this id for pagination", required=false)
@QueryParam("last_id") Long lastId,
@ApiParam(value="number of workflows to return", required=false)
@QueryParam("count") Integer count)
@QueryParam("count") Integer count,
@ApiParam(value="name pattern to be partially matched", required=false)
@QueryParam("name_pattern") String namePattern
)
throws ResourceNotFoundException, AccessControlException
{
final SiteTarget siteTarget = SiteTarget.of(getSiteId());
Expand All @@ -123,6 +126,7 @@ public RestWorkflowDefinitionCollection getWorkflowDefinitions(
List<StoredWorkflowDefinitionWithProject> defs =
rm.getProjectStore(getSiteId())
.getLatestActiveWorkflowDefinitions(Optional.fromNullable(count).or(100), Optional.fromNullable(lastId), // check NotFound first
Optional.fromNullable(namePattern),
ac.getListWorkflowsFilterOfSite(
SiteTarget.of(getSiteId()),
getAuthenticatedUser()));
Expand Down