Skip to content

Commit ead480b

Browse files
phixMephixwslulciuc
authored
OpenSearch indexing and queries (#2834)
* Elasticsearch code. * Adding basic responses for elasticsearch. * Saving highlights. * Saving code cleanup. * Adding EsSearch. * Saving partial progress. * Refinements. * Small bug fixes. * Fixing alignment * Migrating es jobs naming to be specific. * Adding boilerplate for dataset es search * Adding datasets. * Adding polish for more data. * Empty state and other small enhancements. * Adding arrow key functionality. * Removing console log * Spotless * Refinements to queries. * Adding debounce. * Fixing alignment issues. * Saving updates for password setting via env config for elasticsearch. * Setting up startup scripts and adding corresponding waits. * Adding logs and more fields for jobs. * Resolving jackson serialization issue. * Small updates for search display. * Adding onClick handlers. * Fixing null cases, adding more search options for datasets. * Handling enter key. * Fixing minor encoding and layout issues for spark related open lineage events. * Additional fixes for text overflow on names and namespaces. * Fixing indexing problem. * Transitioning to opensearch. * Removing elasticsearch references. * Isolation of search code, calling services. * Adding config to support multiple instances. * Spotless * Adding helm files. * Adding in stronger password for search. * Handling debouncing. * Adding "ADVANCED_SEARCH" configurable variable for web. * Fixing some tests. * Moving indexing down a row. * Spotless * Putting back removed code. * Merge spotless resolution. * Skipping over search for db migration tests. * Adding search back to migration * Trying out ci config setting. * Removing search from base config as a whole. * Pushing out header updates. * Review comment on search service init. * Fixing up dependencies in docker to apply migrations. * Code review updates and naming changes. * newline * Updating for beta vs. non beta endpoints in search resource. * Moving search resource to its own place. * Removing prints. * Removing all helm changes for this work stream. * Adding back lock file contents. * Adding header * Adding middleware proxy. * Code review updates. * Moving from outer gradle to api gradle. * Removing extra containers. * Removing extra containers. * Set timeout for seed container to 60s Signed-off-by: wslulciuc <willy@datakin.com> * Fixing `--no-search` and frontend config. * Add check before indexing ol event Signed-off-by: wslulciuc <willy@datakin.com> * Fix db migration CI job Signed-off-by: wslulciuc <willy@datakin.com> --------- Signed-off-by: wslulciuc <willy@datakin.com> Co-authored-by: phix <peter.hicks@astronomer.io> Co-authored-by: wslulciuc <willy@datakin.com>
1 parent 422fd43 commit ead480b

46 files changed

Lines changed: 1591 additions & 288 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎.circleci/db-migration.sh‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,7 @@ if ! ./docker/up.sh \
6464
--args "--exit-code-from seed_marquez" \
6565
--tag "${MARQUEZ_VERSION}" \
6666
--no-web \
67+
--no-search \
6768
--seed > /dev/null; then
6869
error "failed to start db using backup!"
6970
exit_with_cause
@@ -77,6 +78,7 @@ log "start db using backup (marquez=${MARQUEZ_BUILD_VERSION}):"
7778
if ! ./docker/up.sh \
7879
--args "--exit-code-from seed_marquez" \
7980
--no-web \
81+
--no-search \
8082
--no-volumes \
8183
--build \
8284
--seed > /dev/null; then

‎.env.example‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,4 +2,5 @@ API_PORT=5000
22
API_ADMIN_PORT=5001
33
WEB_PORT=3000
44
POSTGRES_PORT=5432
5+
SEARCH_PORT=9200
56
TAG=0.49.0

‎api/build.gradle‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,9 @@ dependencies {
5151
implementation 'com.graphql-java:graphql-java:20.9'
5252
implementation 'com.graphql-java-kickstart:graphql-java-servlet:12.0.0'
5353

54+
implementation 'org.opensearch.client:opensearch-rest-client:2.15.0'
55+
implementation 'org.opensearch.client:opensearch-java:2.6.0'
56+
5457
testImplementation "io.dropwizard:dropwizard-testing:${dropwizardVersion}"
5558
testImplementation "org.jdbi:jdbi3-testing:${jdbi3Version}"
5659
testImplementation "org.jdbi:jdbi3-testcontainers:${jdbi3Version}"

‎api/src/main/java/marquez/MarquezApp.java‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -137,7 +137,11 @@ public void run(@NonNull MarquezConfig config, @NonNull Environment env) {
137137

138138
final Jdbi jdbi = newJdbi(config, env, source);
139139
final MarquezContext marquezContext =
140-
MarquezContext.builder().jdbi(jdbi).tags(config.getTags()).build();
140+
MarquezContext.builder()
141+
.jdbi(jdbi)
142+
.searchConfig(config.getSearchConfig())
143+
.tags(config.getTags())
144+
.build();
141145

142146
registerResources(config, env, marquezContext);
143147
registerServlets(env);

‎api/src/main/java/marquez/MarquezConfig.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
import marquez.db.FlywayFactory;
1717
import marquez.graphql.GraphqlConfig;
1818
import marquez.jobs.DbRetentionConfig;
19+
import marquez.search.SearchConfig;
1920
import marquez.service.models.Tag;
2021
import marquez.tracing.SentryConfig;
2122

@@ -44,6 +45,10 @@ public class MarquezConfig extends Configuration {
4445
@JsonProperty("sentry")
4546
private final SentryConfig sentry = new SentryConfig();
4647

48+
@Getter
49+
@JsonProperty("search")
50+
private final SearchConfig searchConfig = new SearchConfig();
51+
4752
@Getter
4853
@Setter
4954
@JsonProperty("dbRetention")

‎api/src/main/java/marquez/MarquezContext.java‎

Lines changed: 19 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,7 @@
4343
import marquez.db.TagDao;
4444
import marquez.graphql.GraphqlSchemaBuilder;
4545
import marquez.graphql.MarquezGraphqlServletBuilder;
46+
import marquez.search.SearchConfig;
4647
import marquez.service.ColumnLineageService;
4748
import marquez.service.DatasetFieldService;
4849
import marquez.service.DatasetService;
@@ -53,6 +54,7 @@
5354
import marquez.service.OpenLineageService;
5455
import marquez.service.RunService;
5556
import marquez.service.RunTransitionListener;
57+
import marquez.service.SearchService;
5658
import marquez.service.ServiceFactory;
5759
import marquez.service.SourceService;
5860
import marquez.service.TagService;
@@ -89,26 +91,31 @@ public final class MarquezContext {
8991
@Getter private final OpenLineageService openLineageService;
9092
@Getter private final LineageService lineageService;
9193
@Getter private final ColumnLineageService columnLineageService;
94+
@Getter private final SearchService searchService;
9295
@Getter private final NamespaceResource namespaceResource;
9396
@Getter private final SourceResource sourceResource;
9497
@Getter private final DatasetResource datasetResource;
9598
@Getter private final ColumnLineageResource columnLineageResource;
9699
@Getter private final JobResource jobResource;
97100
@Getter private final TagResource tagResource;
98101
@Getter private final OpenLineageResource openLineageResource;
102+
@Getter private final marquez.api.v2beta.SearchResource v2BetasearchResource;
99103
@Getter private final SearchResource searchResource;
100104
@Getter private final ImmutableList<Object> resources;
101105
@Getter private final JdbiExceptionExceptionMapper jdbiException;
102106
@Getter private final JsonProcessingExceptionMapper jsonException;
103107
@Getter private final GraphQLHttpServlet graphqlServlet;
108+
@Getter private final SearchConfig searchConfig;
104109

105110
private MarquezContext(
106111
@NonNull final Jdbi jdbi,
112+
@NonNull final SearchConfig searchConfig,
107113
@NonNull final ImmutableSet<Tag> tags,
108114
List<RunTransitionListener> runTransitionListeners) {
109115
if (runTransitionListeners == null) {
110116
runTransitionListeners = new ArrayList<>();
111117
}
118+
this.searchConfig = searchConfig;
112119

113120
final BaseDao baseDao = jdbi.onDemand(NamespaceDao.class);
114121
this.namespaceDao = jdbi.onDemand(NamespaceDao.class);
@@ -141,6 +148,7 @@ private MarquezContext(
141148
this.openLineageService = new OpenLineageService(baseDao, runService);
142149
this.lineageService = new LineageService(lineageDao, jobDao);
143150
this.columnLineageService = new ColumnLineageService(columnLineageDao, datasetFieldDao);
151+
this.searchService = new SearchService(searchConfig);
144152
this.jdbiException = new JdbiExceptionExceptionMapper();
145153
this.jsonException = new JsonProcessingExceptionMapper();
146154
final ServiceFactory serviceFactory =
@@ -151,6 +159,7 @@ private MarquezContext(
151159
.namespaceService(namespaceService)
152160
.tagService(tagService)
153161
.openLineageService(openLineageService)
162+
.searchService(searchService)
154163
.sourceService(sourceService)
155164
.lineageService(lineageService)
156165
.columnLineageService(columnLineageService)
@@ -165,6 +174,7 @@ private MarquezContext(
165174
this.tagResource = new TagResource(serviceFactory);
166175
this.openLineageResource = new OpenLineageResource(serviceFactory, openLineageDao);
167176
this.searchResource = new SearchResource(searchDao);
177+
this.v2BetasearchResource = new marquez.api.v2beta.SearchResource(serviceFactory);
168178

169179
this.resources =
170180
ImmutableList.of(
@@ -177,7 +187,8 @@ private MarquezContext(
177187
jdbiException,
178188
jsonException,
179189
openLineageResource,
180-
searchResource);
190+
searchResource,
191+
v2BetasearchResource);
181192

182193
final MarquezGraphqlServletBuilder servlet = new MarquezGraphqlServletBuilder();
183194
this.graphqlServlet = servlet.getServlet(new GraphqlSchemaBuilder(jdbi));
@@ -190,6 +201,7 @@ public static Builder builder() {
190201
public static class Builder {
191202

192203
private Jdbi jdbi;
204+
private SearchConfig searchConfig;
193205
private ImmutableSet<Tag> tags;
194206
private List<RunTransitionListener> runTransitionListeners;
195207

@@ -203,6 +215,11 @@ public Builder jdbi(@NonNull Jdbi jdbi) {
203215
return this;
204216
}
205217

218+
public Builder searchConfig(@NonNull SearchConfig searchConfig) {
219+
this.searchConfig = searchConfig;
220+
return this;
221+
}
222+
206223
public Builder tags(@NonNull ImmutableSet<Tag> tags) {
207224
this.tags = tags;
208225
return this;
@@ -219,7 +236,7 @@ public Builder runTransitionListeners(
219236
}
220237

221238
public MarquezContext build() {
222-
return new MarquezContext(jdbi, tags, runTransitionListeners);
239+
return new MarquezContext(jdbi, searchConfig, tags, runTransitionListeners);
223240
}
224241
}
225242
}

‎api/src/main/java/marquez/api/OpenLineageResource.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,9 @@ public OpenLineageResource(
6868
public void create(@Valid @NotNull BaseEvent event, @Suspended final AsyncResponse asyncResponse)
6969
throws JsonProcessingException, SQLException {
7070
if (event instanceof LineageEvent) {
71+
if (serviceFactory.getSearchService().isEnabled()) {
72+
serviceFactory.getSearchService().indexEvent((LineageEvent) event);
73+
}
7174
openLineageService
7275
.createAsync((LineageEvent) event)
7376
.whenComplete((result, err) -> onComplete(result, err, asyncResponse));

‎api/src/main/java/marquez/api/v2beta/SearchResource.java‎

Lines changed: 91 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,91 @@
1+
/*
2+
* Copyright 2018-2024 contributors to the Marquez project
3+
* SPDX-License-Identifier: Apache-2.0
4+
*/
5+
6+
package marquez.api.v2beta;
7+
8+
import static javax.ws.rs.core.MediaType.APPLICATION_JSON;
9+
10+
import com.codahale.metrics.annotation.ExceptionMetered;
11+
import com.codahale.metrics.annotation.ResponseMetered;
12+
import com.codahale.metrics.annotation.Timed;
13+
import com.fasterxml.jackson.annotation.JsonCreator;
14+
import com.fasterxml.jackson.databind.node.ObjectNode;
15+
import java.io.IOException;
16+
import java.util.List;
17+
import java.util.Map;
18+
import java.util.stream.Collectors;
19+
import javax.validation.constraints.NotBlank;
20+
import javax.ws.rs.GET;
21+
import javax.ws.rs.Path;
22+
import javax.ws.rs.Produces;
23+
import javax.ws.rs.QueryParam;
24+
import javax.ws.rs.core.Response;
25+
import lombok.Getter;
26+
import lombok.NonNull;
27+
import lombok.ToString;
28+
import lombok.extern.slf4j.Slf4j;
29+
import marquez.service.SearchService;
30+
import marquez.service.ServiceFactory;
31+
import org.opensearch.client.opensearch.core.SearchResponse;
32+
import org.opensearch.client.opensearch.core.search.Hit;
33+
34+
@Slf4j
35+
@Path("/api/v2beta/search")
36+
public class SearchResource {
37+
38+
private final SearchService searchService;
39+
40+
public SearchResource(@NonNull final ServiceFactory serviceFactory) {
41+
this.searchService = serviceFactory.getSearchService();
42+
}
43+
44+
@Timed
45+
@ResponseMetered
46+
@ExceptionMetered
47+
@GET
48+
@Produces(APPLICATION_JSON)
49+
@Path("jobs")
50+
public Response searchJobs(@QueryParam("q") @NotBlank String query) throws IOException {
51+
if (searchService.isEnabled()) {
52+
return Response.status(Response.Status.SERVICE_UNAVAILABLE).build();
53+
}
54+
return formatOpenSearchResponse(this.searchService.searchJobs(query));
55+
}
56+
57+
@Timed
58+
@ResponseMetered
59+
@ExceptionMetered
60+
@GET
61+
@Produces(APPLICATION_JSON)
62+
@Path("datasets")
63+
public Response searchDatasets(@QueryParam("q") @NotBlank String query) throws IOException {
64+
if (searchService.isEnabled()) {
65+
return Response.status(Response.Status.SERVICE_UNAVAILABLE).build();
66+
}
67+
return formatOpenSearchResponse(this.searchService.searchDatasets(query));
68+
}
69+
70+
private Response formatOpenSearchResponse(SearchResponse<ObjectNode> response) {
71+
List<ObjectNode> hits =
72+
response.hits().hits().stream().map(Hit::source).collect(Collectors.toList());
73+
List<Map<String, List<String>>> highlights =
74+
response.hits().hits().stream().map(Hit::highlight).collect(Collectors.toList());
75+
76+
return Response.ok(new OpenSearchResult(hits, highlights)).build();
77+
}
78+
79+
@ToString
80+
public static final class OpenSearchResult {
81+
@Getter private final List<ObjectNode> hits;
82+
@Getter private final List<Map<String, List<String>>> highlights;
83+
84+
@JsonCreator
85+
public OpenSearchResult(
86+
@NonNull List<ObjectNode> hits, @NonNull List<Map<String, List<String>>> highlights) {
87+
this.hits = hits;
88+
this.highlights = highlights;
89+
}
90+
}
91+
}

‎api/src/main/java/marquez/search/SearchConfig.java‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,30 @@
1+
/*
2+
* Copyright 2018-2024 contributors to the Marquez project
3+
* SPDX-License-Identifier: Apache-2.0
4+
*/
5+
6+
package marquez.search;
7+
8+
import com.fasterxml.jackson.annotation.JsonProperty;
9+
import lombok.Getter;
10+
11+
public class SearchConfig {
12+
public static final boolean ENABLED = false;
13+
public static final String SCHEME = "http";
14+
public static final String HOST = "opensearch";
15+
public static final int PORT = 9200;
16+
public static final String USERNAME = "admin";
17+
public static final String PASSWORD = "admin";
18+
19+
@Getter @JsonProperty private boolean enabled = ENABLED;
20+
21+
@Getter @JsonProperty private String scheme = SCHEME;
22+
23+
@Getter @JsonProperty private String host = HOST;
24+
25+
@Getter @JsonProperty private int port = PORT;
26+
27+
@Getter @JsonProperty private String username = USERNAME;
28+
29+
@Getter @JsonProperty private String password = PASSWORD;
30+
}

0 commit comments

Comments
 (0)
⚡