| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more |
| * contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. |
| * The ASF licenses this file to You under the Apache License, Version 2.0 |
| * (the "License"); you may not use this file except in compliance with |
| * the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| package org.apache.unomi.shell.migration.impl; |
| |
| import org.apache.http.HttpStatus; |
| import org.apache.http.client.methods.CloseableHttpResponse; |
| import org.apache.http.client.methods.HttpEntityEnclosingRequestBase; |
| import org.apache.http.client.methods.HttpGet; |
| import org.apache.http.client.methods.HttpPost; |
| import org.apache.http.client.methods.HttpPut; |
| import org.apache.http.entity.StringEntity; |
| import org.apache.http.impl.client.CloseableHttpClient; |
| import org.apache.http.util.EntityUtils; |
| import org.apache.karaf.shell.api.console.Session; |
| import org.apache.unomi.shell.migration.Migration; |
| import org.apache.unomi.shell.migration.MigrationConfig; |
| import org.apache.unomi.shell.migration.utils.ConsoleUtils; |
| import org.apache.unomi.shell.migration.utils.MigrationUtils; |
| import org.json.JSONArray; |
| import org.json.JSONObject; |
| import org.osgi.framework.BundleContext; |
| |
| import java.io.IOException; |
| import java.net.URI; |
| import java.time.Instant; |
| import java.util.Collections; |
| import java.util.HashSet; |
| import java.util.Map; |
| import java.util.Set; |
| import java.util.stream.Collectors; |
| import java.util.stream.StreamSupport; |
| |
| public class MigrationTo200 implements Migration { |
| |
| private CloseableHttpClient httpClient; |
| private Session session; |
| private String esAddress; |
| private String indexPrefix; |
| private BundleContext bundleContext; |
| private MigrationConfig migrationConfig; |
| |
| @Override |
| public void execute(Session session, CloseableHttpClient httpClient, MigrationConfig migrationConfig, BundleContext bundleContext) throws IOException { |
| this.httpClient = httpClient; |
| this.session = session; |
| this.esAddress = migrationConfig.getString(MigrationConfig.CONFIG_ES_ADDRESS, session); |
| this.indexPrefix = migrationConfig.getString(MigrationConfig.INDEX_PREFIX, session); |
| this.bundleContext = bundleContext; |
| this.migrationConfig = migrationConfig; |
| |
| doExecute(); |
| } |
| |
| private void doExecute() throws IOException { |
| Set<String> indexes = MigrationUtils.getIndexesPrefixedBy(httpClient, esAddress, indexPrefix + "-event-"); |
| createScopeMapping(); |
| createScopes(getSetOfScopes(indexes)); |
| createProfileAliasDocumentsFromProfile(); |
| } |
| |
| private boolean scopeIndexNotExists() throws IOException { |
| final HttpGet httpGet = new HttpGet(esAddress + "/" + indexPrefix + "-scope"); |
| |
| httpGet.addHeader("Accept", "application/json"); |
| httpGet.addHeader("Content-Type", "application/json"); |
| |
| try (CloseableHttpResponse response = httpClient.execute(httpGet)) { |
| return response.getStatusLine().getStatusCode() != HttpStatus.SC_OK; |
| } |
| } |
| |
| private void createScopeMapping() throws IOException { |
| |
| if (scopeIndexNotExists()) { |
| System.out.println("Creation for index = \"" + indexPrefix + "-scope\" starting."); |
| final HttpPut httpPost = new HttpPut(esAddress + "/" + indexPrefix + "-scope"); |
| |
| httpPost.addHeader("Accept", "application/json"); |
| httpPost.addHeader("Content-Type", "application/json"); |
| |
| String baseRequest = MigrationUtils.resourceAsString(bundleContext,"requestBody/2.0.0/base_index_mapping.json"); |
| String mapping = MigrationUtils.extractMappingFromBundles(bundleContext, "scope.json"); |
| // We intentionally extract setting from profile index, because the scope index doesnt exist yet, and all indices share the same configuration regarding shards, replicas, etc .. |
| String request = MigrationUtils.buildIndexCreationRequest(httpClient, esAddress, baseRequest, indexPrefix + "-profile", mapping); |
| |
| httpPost.setEntity(new StringEntity(request)); |
| |
| try (CloseableHttpResponse response = httpClient.execute(httpPost)) { |
| if (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) { |
| System.out.println(indexPrefix + "-scope has been correctly created"); |
| } else { |
| System.out.println( |
| "Failed to create the index " + indexPrefix + "-scope.Code:" + response.getStatusLine().getStatusCode()); |
| throw new RuntimeException("Can not create the scope index. Stop the execution of the migration."); |
| } |
| } |
| } else { |
| System.out.println("The scope index already exists. Skipping the creation of this index"); |
| } |
| |
| } |
| |
| private void createScopes(Set<String> scopes) throws IOException { |
| final StringBuilder body = new StringBuilder(); |
| String saveScopeBody = MigrationUtils.resourceAsString(bundleContext,"requestBody/bulkSaveScope.ndjson"); |
| scopes.forEach(scope -> body.append(saveScopeBody.replace("$scope", scope))); |
| |
| final HttpPost httpPost = new HttpPost(esAddress + "/" + indexPrefix + "-scope/_bulk"); |
| |
| httpPost.addHeader("Accept", "application/json"); |
| httpPost.addHeader("Content-Type", "application/x-ndjson"); |
| |
| httpPost.setEntity(new StringEntity(body.toString())); |
| |
| try (CloseableHttpResponse response = httpClient.execute(httpPost)) { |
| if (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) { |
| System.out.println("Creating the \"scopes\" into the index " + indexPrefix + "-scope successfully finished"); |
| } else { |
| System.out.println("Creating the \"scopes\" into the index " + indexPrefix + "-scope has failed" + response.getStatusLine() |
| .getStatusCode()); |
| } |
| } |
| } |
| |
| private Set<String> getSetOfScopes(Set<String> indices) throws IOException { |
| String joinedIndices = String.join(",", indices); |
| final HttpPost httpPost = new HttpPost(esAddress + "/" + joinedIndices + "/_search"); |
| |
| httpPost.addHeader("Accept", "application/json"); |
| httpPost.addHeader("Content-Type", "application/json"); |
| |
| String request = MigrationUtils.resourceAsString(bundleContext,"requestBody/searchScope.json"); |
| |
| httpPost.setEntity(new StringEntity(request)); |
| |
| Set<String> scopes = new HashSet<>(); |
| try (CloseableHttpResponse response = httpClient.execute(httpPost)) { |
| JSONObject responseAsJson = new JSONObject(EntityUtils.toString(response.getEntity())); |
| if (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) { |
| System.out.println("Getting the \"scope\" values from the events successfully finished. " + "Number of scope to create: " |
| + responseAsJson.getJSONObject("aggregations").getJSONObject("bucketInfos").get("count").toString()); |
| scopes = StreamSupport |
| .stream(responseAsJson.getJSONObject("aggregations").getJSONObject("scopes").getJSONArray("buckets").spliterator(), |
| false).map(bucketElement -> ((JSONObject) bucketElement).getString("key")).collect(Collectors.toSet()); |
| } else { |
| System.out.println( |
| "Getting the \"scope\" values from the event has failed. Code: " + response.getStatusLine().getStatusCode()); |
| } |
| } |
| return scopes; |
| } |
| |
| private void createProfileAliasDocumentsFromProfile() throws IOException { |
| System.out.println("Migration \"Create profileAlias from profile\" started"); |
| Instant migrationTime = Instant.now(); |
| int size = 1000; |
| doProcessProfiles(migrationTime, size); |
| System.out.println("Migration \"Create profileAlias from profile\" completed."); |
| } |
| |
| private void doProcessProfiles(Instant migrationTime, int offset) throws IOException { |
| CloseableHttpResponse response = null; |
| try { |
| response = httpClient.execute(createSearchRequest(offset)); |
| |
| while (true) { |
| JSONObject responseAsJson = getResponseAsJSON(response); |
| |
| String scrollId = responseAsJson.has("_scroll_id") ? responseAsJson.getString("_scroll_id"): null; |
| JSONArray hits = getProfileHits(responseAsJson); |
| |
| if (hits.length() == 0) { |
| if (scrollId != null) { |
| CloseableHttpResponse deleteScrollResponse = httpClient.execute(createDeleteScrollRequest(scrollId)); |
| if (deleteScrollResponse != null) { |
| deleteScrollResponse.close(); |
| } |
| } |
| break; |
| } |
| |
| StringBuilder bulkCreateRequest = new StringBuilder(); |
| for (Object o : hits) { |
| JSONObject hit = (JSONObject) o; |
| if (hit.has("_source")) { |
| JSONObject profile = hit.getJSONObject("_source"); |
| if (profile.has("itemId")) { |
| String itemId = profile.getString("itemId"); |
| String bulkSaveProfileAliases = MigrationUtils.resourceAsString(bundleContext,"requestBody/bulkSaveProfileAliases.ndjson"); |
| bulkCreateRequest.append(bulkSaveProfileAliases. |
| replace("$itemId", itemId). |
| replace("$migrationTime", migrationTime.toString())); |
| } |
| } |
| } |
| |
| CloseableHttpResponse bulkResponse = httpClient.execute(createProfileAliasRequest(bulkCreateRequest.toString())); |
| if (bulkResponse != null) { |
| bulkResponse.close(); |
| } |
| |
| response = httpClient.execute(createSearchRequestWithScrollId(scrollId)); |
| } |
| } finally { |
| if (response != null) { |
| response.close(); |
| } |
| } |
| } |
| |
| private JSONObject getResponseAsJSON(CloseableHttpResponse response) throws IOException { |
| if (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) { |
| return new JSONObject(EntityUtils.toString(response.getEntity())); |
| } |
| return new JSONObject(); |
| } |
| |
| private JSONArray getProfileHits(JSONObject responseAsJson) { |
| if (responseAsJson.has("hits")) { |
| JSONObject hitsObject = responseAsJson.getJSONObject("hits"); |
| if (hitsObject.has("hits")) { |
| return hitsObject.getJSONArray("hits"); |
| } |
| } |
| return new JSONArray(); |
| } |
| |
| private HttpPost createSearchRequestWithScrollId(final String scrollId) throws IOException { |
| final String requestBody = "{\n" + |
| " \"scroll_id\": \"" + scrollId + "\",\n" + |
| " \"scroll\": \"1h\"\n" + |
| "}"; |
| |
| final HttpPost request = new HttpPost(esAddress + "/_search/scroll"); |
| |
| request.addHeader("Accept", "application/json"); |
| request.addHeader("Content-Type", "application/json"); |
| request.setEntity(new StringEntity(requestBody)); |
| |
| return request; |
| } |
| |
| private HttpGet createSearchRequest(int size) { |
| return new HttpGet(esAddress + "/context-profile/_search?&scroll=1h&_source_includes=itemId&size=" + size); |
| } |
| |
| private HttpEntityEnclosingRequestBase createDeleteScrollRequest(final String scrollId) throws IOException { |
| final HttpEntityEnclosingRequestBase deleteRequest = new HttpEntityEnclosingRequestBase() { |
| @Override |
| public String getMethod() { |
| return "DELETE"; |
| } |
| }; |
| |
| deleteRequest.setURI(URI.create(esAddress + "/_search/scroll")); |
| deleteRequest.setEntity(new StringEntity("{ \"scroll_id\": \"" + scrollId + "\" }")); |
| deleteRequest.addHeader("Accept", "application/json"); |
| deleteRequest.addHeader("Content-Type", "application/json"); |
| |
| return deleteRequest; |
| } |
| |
| private HttpPost createProfileAliasRequest(String bulkRequestAsString) throws IOException { |
| final HttpPost bulkRequest = new HttpPost(esAddress + "/context-profilealias/_bulk"); |
| |
| bulkRequest.addHeader("Accept", "application/json"); |
| bulkRequest.addHeader("Content-Type", "application/json"); |
| bulkRequest.setEntity(new StringEntity(bulkRequestAsString)); |
| |
| return bulkRequest; |
| } |
| } |