This is an automated email from the ASF dual-hosted git repository.
fjohn pushed a commit to branch datalake-rest-extension
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
The following commit(s) were added to refs/heads/datalake-rest-extension by
this push:
new 766b0d5 Changes to DeleteMeasurement + New
WriteMeasurementFromDefaultToCustomPolicy
766b0d5 is described below
commit 766b0d5685051599efe61f3a38d946ae5d35c8c2
Author: Felix John <[email protected]>
AuthorDate: Fri Sep 25 10:55:33 2020 +0200
Changes to DeleteMeasurement + New WriteMeasurementFromDefaultToCustomPolicy
---
.../rest/impl/datalake/DataLakeManagementV3.java | 39 +++++++++++-----------
.../rest/impl/datalake/DataLakeResourceV3.java | 9 +++++
2 files changed, 29 insertions(+), 19 deletions(-)
diff --git
a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/datalake/DataLakeManagementV3.java
b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/datalake/DataLakeManagementV3.java
index 4f6b860..5a99e0f 100644
---
a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/datalake/DataLakeManagementV3.java
+++
b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/datalake/DataLakeManagementV3.java
@@ -239,7 +239,6 @@ public class DataLakeManagementV3 {
if (rp == null) rp =
BackendConfig.INSTANCE.getDefaultRetentionPolicyName();
-
Query query = new Query("SELECT * FROM "
+ BackendConfig.INSTANCE.getInfluxDatabaseName() + "." + rp + "."
+ index
+ " ORDER BY time LIMIT " + itemsPerPage + " OFFSET " + page *
itemsPerPage,
@@ -254,6 +253,22 @@ public class DataLakeManagementV3 {
return new PageResult(dataResult.getTotal(), dataResult.getHeaders(),
dataResult.getRows(), page, pageSum);
}
+ public void writeMeasurementFromDefaulToCustomRententionPolicy(String index,
String rp) {
+ InfluxDB influxDB = getInfluxDBClient();
+ String rp_default = BackendConfig.INSTANCE.getDefaultRetentionPolicyName();
+
+ QueryResult result = influxDB.query(new Query("SELECT * INTO " +
+ BackendConfig.INSTANCE.getInfluxDatabaseName() + "." + rp + "." +
index + "_temp"
+ + " FROM " + BackendConfig.INSTANCE.getInfluxDatabaseName() + "."
+ rp_default + "." + index));
+ deleteMeasurement(index);
+
+ QueryResult result_2 = influxDB.query(new Query("SELECT * INTO " +
+ BackendConfig.INSTANCE.getInfluxDatabaseName() + "." + rp + "." +
index
+ + " FROM " + BackendConfig.INSTANCE.getInfluxDatabaseName() + "."
+ rp + "." + index + "_temp"));
+ deleteMeasurement(index + "_temp");
+ influxDB.close();
+
+ }
public DataResult getRetentionPoliciesOfDatabase() {
InfluxDB influxDB = getInfluxDBClient();
@@ -276,9 +291,7 @@ public class DataLakeManagementV3 {
if (shardDuration != null) { shardDurationString = " SHARD DURATION " +
shardDuration; }
Query query = new Query("CREATE RETENTION POLICY "
- + rp
- + " ON "
- + BackendConfig.INSTANCE.getInfluxDatabaseName()
+ + rp + " ON " + BackendConfig.INSTANCE.getInfluxDatabaseName()
+ durationString
+ replicationString
+ shardDurationString,
@@ -305,9 +318,7 @@ public class DataLakeManagementV3 {
if (shardDuration != null) { shardDurationString = " SHARD DURATION " +
shardDuration; }
Query query = new Query("ALTER RETENTION POLICY "
- + rp
- + " ON "
- + BackendConfig.INSTANCE.getInfluxDatabaseName()
+ + rp + " ON " + BackendConfig.INSTANCE.getInfluxDatabaseName()
+ durationString
+ replicationString
+ shardDurationString,
@@ -381,24 +392,15 @@ public class DataLakeManagementV3 {
return storageSize;
}
- public static void main(String [] args) {
- DataLakeManagementV3 dlmv3 = new DataLakeManagementV3();
- InfluxDB influxDB = dlmv3.getInfluxDBClient();
- long size = dlmv3.getStorageSizeOfDatabase();
- System.out.println(size);
- // dlmv3.deleteRetentionPolicy("rp10");
- // DataResult result = dlmv3.getRetentionPoliciesOfDatabase();
- // dlmv3.createRetentionPolicy("rp8", "1h", null, 5, Boolean.FALSE);
- System.out.println(dlmv3.getNumOfRecordsOfTable("felix4", null));
- }
-
public void deleteMeasurement(String index) {
InfluxDB influxDB = getInfluxDBClient();
+ influxDB.setDatabase(BackendConfig.INSTANCE.getInfluxDatabaseName());
Query query = new Query("DROP MEASUREMENT " + index);
QueryResult influx_result = influxDB.query(query);
if (influx_result.hasError() ||
influx_result.getResults().get(0).getError() != null) {
+ System.out.println(influx_result.getResults().get(0).getError());
System.out.println("Error!");
// TODO: Raise error exception
}
@@ -419,7 +421,6 @@ public class DataLakeManagementV3 {
}
couchDbClient.shutdown();
-
// TODO: Raise exception if both not successful
}
diff --git
a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/datalake/DataLakeResourceV3.java
b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/datalake/DataLakeResourceV3.java
index 9734cf5..0f11b7f 100644
---
a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/datalake/DataLakeResourceV3.java
+++
b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/datalake/DataLakeResourceV3.java
@@ -276,6 +276,15 @@ public class DataLakeResourceV3 extends
AbstractRestInterface {
}
@GET
+ @Path("/data/{index}/default/{name}")
+ @Produces(MediaType.TEXT_PLAIN)
+ public Response
writeMeasurementFromDefaulToCustomRententionPolicy(@PathParam("index") String
index,
+ @PathParam("name") String policyRetentionName) {
+
this.dataLakeManagement.writeMeasurementFromDefaulToCustomRententionPolicy(index,
policyRetentionName);
+ return Response.ok("Wrote measurement to other retention policy.",
MediaType.TEXT_PLAIN).build();
+ }
+
+ @GET
@Produces(MediaType.TEXT_PLAIN)
@Path("/database/size")
public Response getStorageSizeOfDatabase() {
