dev cloud edge policy

conf/config.properties

@@ -15,8 +15,7 @@ username=root
 password=root
 
 # 时序数据库列表，使用','分隔不同实例
-storageEngineList=127.0.0.1#6667#iotdb#username=root#password=root#sessionPoolSize=100#dataDir=/path/to/your/data/
-#storageEngineList=127.0.0.1#8086#influxdb#url=http://localhost:8086/#token=your-token#organization=your-organization
+storageEngineList=127.0.0.1#6667#iotdb#username=root#password=root#sessionPoolSize=30#edgeName=edge1
 
 # 异步请求最大重复次数
 maxAsyncRetryTimes=3
@@ -34,13 +33,7 @@ replicaNum=1
 databaseClassNames=iotdb=cn.edu.tsinghua.iginx.iotdb.IoTDBPlanExecutor,influxdb=cn.edu.tsinghua.iginx.influxdb.InfluxDBPlanExecutor
 
 # 策略类名
-policyClassName=cn.edu.tsinghua.iginx.policy.naive.NativePolicy
-
-# 统计信息收集类
-# statisticsCollectorClassName=cn.edu.tsinghua.iginx.statistics.StatisticsCollector
-
-# 统计信息打印间隔，单位毫秒
-# statisticsLogInterval=1000
+policyClassName=cn.edu.tsinghua.iginx.policy.cloud.EdgeCloudCollaborationPolicy
 
 ####################
 ### Rest 服务配置
@@ -53,7 +46,7 @@ restIp=0.0.0.0
 restPort=6666
 
 # 是否启用 rest 服务
-enableRestService=true
+enableRestService=false
 
 # 乱序数据 margin, 单位是秒
 disorderMargin=10
@@ -87,6 +80,23 @@ mqtt_port=1883
 
 mqtt_handler_pool_size=1
 
-mqtt_payload_formatter=cn.edu.tsinghua.iginx.mqtt.JsonPayloadFormatter
+mqtt_payload_formatter=cn.edu.tsinghua.iginx.mqtt.ShangFeiPayloadFormatter
 
 mqtt_max_message_size=1048576
+
+####################
+### 边云协同 配置
+####################
+
+# 是否开启边云协同功能，默认关闭
+enable_edge_cloud_collaboration=true
+
+# iginx 是否为边缘端，默认为非边缘端
+is_edge=true
+
+# 边缘端名字，所有由边缘端写入的序列均包含该前缀；非边缘端不需要填写
+<<<<<<< HEAD
+edge_name=
+=======
+edge_name=edge2
+>>>>>>> temp storage

core/src/main/java/cn/edu/tsinghua/iginx/conf/Config.java

@@ -78,6 +78,12 @@ public class Config {
 
     private int mqttMaxMessageSize = 1048576;
 
+    private boolean enableEdgeCloudCollaboration = false;
+
+    private boolean isEdge = false;
+
+    private String edgeName = "";
+
     public int getMaxTimeseriesLength() {
         return maxTimeseriesLength;
     }
@@ -309,4 +315,28 @@ public int getMqttMaxMessageSize() {
     public void setMqttMaxMessageSize(int mqttMaxMessageSize) {
         this.mqttMaxMessageSize = mqttMaxMessageSize;
     }
+
+    public boolean isEnableEdgeCloudCollaboration() {
+        return enableEdgeCloudCollaboration;
+    }
+
+    public void setEnableEdgeCloudCollaboration(boolean enableEdgeCloudCollaboration) {
+        this.enableEdgeCloudCollaboration = enableEdgeCloudCollaboration;
+    }
+
+    public boolean isEdge() {
+        return isEdge;
+    }
+
+    public void setEdge(boolean edge) {
+        isEdge = edge;
+    }
+
+    public String getEdgeName() {
+        return edgeName;
+    }
+
+    public void setEdgeName(String edgeName) {
+        this.edgeName = edgeName;
+    }
 }

core/src/main/java/cn/edu/tsinghua/iginx/conf/ConfigDescriptor.java

@@ -89,6 +89,10 @@ private void loadPropsFromFile() {
             config.setMqttHandlerPoolSize(Integer.parseInt(properties.getProperty("mqtt_handler_pool_size", "1")));
             config.setMqttPayloadFormatter(properties.getProperty("mqtt_payload_formatter", "cn.edu.tsinghua.iginx.mqtt.JsonPayloadFormatter"));
             config.setMqttMaxMessageSize(Integer.parseInt(properties.getProperty("mqtt_max_message_size", "1048576")));
+
+            config.setEnableEdgeCloudCollaboration(Boolean.parseBoolean(properties.getProperty("enable_edge_cloud_collaboration", "false")));
+            config.setEdge(Boolean.parseBoolean(properties.getProperty("is_edge", "false")));
+            config.setEdgeName(properties.getProperty("edge_name", ""));
         } catch (IOException e) {
             logger.error("Fail to load properties: ", e);
         }

core/src/main/java/cn/edu/tsinghua/iginx/core/context/InsertColumnRecordsContext.java

@@ -18,10 +18,14 @@
  */
 package cn.edu.tsinghua.iginx.core.context;
 
+import cn.edu.tsinghua.iginx.conf.Config;
+import cn.edu.tsinghua.iginx.conf.ConfigDescriptor;
 import cn.edu.tsinghua.iginx.thrift.InsertColumnRecordsReq;
 import lombok.Data;
 import lombok.EqualsAndHashCode;
 
+import java.util.stream.Collectors;
+
 @EqualsAndHashCode(callSuper = true)
 @Data
 public class InsertColumnRecordsContext extends RequestContext {

core/src/main/java/cn/edu/tsinghua/iginx/core/context/InsertRowRecordsContext.java

@@ -18,10 +18,14 @@
  */
 package cn.edu.tsinghua.iginx.core.context;
 
+import cn.edu.tsinghua.iginx.conf.Config;
+import cn.edu.tsinghua.iginx.conf.ConfigDescriptor;
 import cn.edu.tsinghua.iginx.thrift.InsertRowRecordsReq;
 import lombok.Data;
 import lombok.EqualsAndHashCode;
 
+import java.util.stream.Collectors;
+
 @EqualsAndHashCode(callSuper = true)
 @Data
 public class InsertRowRecordsContext extends RequestContext {
@@ -32,4 +36,8 @@ public InsertRowRecordsContext(InsertRowRecordsReq req) {
         super(req.sessionId, ContextType.InsertRowRecords);
         this.req = req;
     }
+
+    public InsertRowRecordsReq getReq() {
+        return req;
+    }
 }

core/src/main/java/cn/edu/tsinghua/iginx/metadata/DefaultMetaManager.java

@@ -18,6 +18,7 @@
  */
 package cn.edu.tsinghua.iginx.metadata;
 
+import cn.edu.tsinghua.iginx.conf.Config;
 import cn.edu.tsinghua.iginx.conf.ConfigDescriptor;
 import cn.edu.tsinghua.iginx.db.StorageEngine;
 import cn.edu.tsinghua.iginx.exceptions.MetaStorageException;
@@ -126,8 +127,12 @@ private void initIginx() throws MetaStorageException {
         for (IginxMeta iginx: storage.loadIginx().values()) {
             cache.addIginx(iginx);
         }
+        Map<String, String> extraParams = new HashMap<>();
+        extraParams.put("enable_edge_cloud_collaboration", Boolean.toString(ConfigDescriptor.getInstance().getConfig().isEnableEdgeCloudCollaboration()));
+        extraParams.put("is_edge", Boolean.toString(ConfigDescriptor.getInstance().getConfig().isEdge()));
+        extraParams.put("edge_name", ConfigDescriptor.getInstance().getConfig().getEdgeName());
         IginxMeta iginx = new IginxMeta(0L, ConfigDescriptor.getInstance().getConfig().getIp(),
-                ConfigDescriptor.getInstance().getConfig().getPort(), null);
+                ConfigDescriptor.getInstance().getConfig().getPort(), extraParams);
         id = storage.registerIginx(iginx);
         SnowFlakeUtils.init(id);
     }

core/src/main/java/cn/edu/tsinghua/iginx/metadata/entity/IginxMeta.java

@@ -18,6 +18,7 @@
  */
 package cn.edu.tsinghua.iginx.metadata.entity;
 
+import java.util.HashMap;
 import java.util.Map;
 
 public final class IginxMeta {
@@ -62,6 +63,8 @@ public int getPort() {
     }
 
     public Map<String, String> getExtraParams() {
+        if (extraParams == null)
+            return new HashMap<>();
         return extraParams;
     }
 }

core/src/main/java/cn/edu/tsinghua/iginx/metadata/entity/StorageEngineMeta.java

@@ -21,6 +21,7 @@
 import cn.edu.tsinghua.iginx.db.StorageEngine;
 
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
@@ -99,6 +100,8 @@ public void setPort(int port) {
     }
 
     public Map<String, String> getExtraParams() {
+        if (extraParams == null)
+            return new HashMap<>();
         return extraParams;
     }
 

core/src/main/java/cn/edu/tsinghua/iginx/metadata/storage/zk/ZooKeeperMetaStorage.java

@@ -434,6 +434,9 @@ private void registerStorageEngineListener() throws Exception {
                         break;
                     }
                     data = event.getData().getData();
+                    if (event.getData().getPath().equals(STORAGE_ENGINE_NODE_PREFIX)) {
+                        break;
+                    }
                     logger.info("storage engine meta updated " + event.getData().getPath());
                     logger.info("storage engine: " + new String(data));
                     storageEngineMeta = JsonUtils.fromJson(data, StorageEngineMeta.class);

core/src/main/java/cn/edu/tsinghua/iginx/mqtt/ShangFeiPayloadFormatter.java

@@ -0,0 +1,92 @@
+/*
+ * 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 cn.edu.tsinghua.iginx.mqtt;
+
+import cn.edu.tsinghua.iginx.conf.Config;
+import cn.edu.tsinghua.iginx.conf.ConfigDescriptor;
+import cn.edu.tsinghua.iginx.thrift.DataType;
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+import com.google.gson.JsonArray;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import io.netty.buffer.ByteBuf;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+public class ShangFeiPayloadFormatter implements IPayloadFormatter {
+
+    private static final Gson gson = new GsonBuilder().create();
+
+    private static final Logger logger = LoggerFactory.getLogger(ShangFeiPayloadFormatter.class);
+
+    private final Config config = ConfigDescriptor.getInstance().getConfig();
+
+    @Override
+    public List<Message> format(ByteBuf payload) {
+        if (payload == null) {
+            return null;
+        }
+        String txt = payload.toString(StandardCharsets.UTF_8);
+        JsonArray jsonArray = gson.fromJson(txt, JsonArray.class);
+        List<Message> messages = new ArrayList<>();
+
+        for (int i = 0; i < jsonArray.size(); i++) {
+            JsonObject jsonObject = jsonArray.get(i).getAsJsonObject();
+            JsonObject metadata = jsonObject.get("metadata").getAsJsonObject();
+            JsonElement value = jsonObject.get("value");
+
+            String displayName = metadata.get("displayName").getAsString();
+            String path = metadata.get("path").getAsString().replace('/', '.').substring(1) + jsonObject.get("name").getAsString() + "@" + displayName;
+            if (config.isEnableEdgeCloudCollaboration() && config.isEdge() && !config.getEdgeName().equals("")) {
+                path = config.getEdgeName() + "." + path;
+            }
+            long timestamp = jsonObject.get("timestamp").getAsLong();
+
+            Message message = new Message();
+            message.setPath(path);
+            message.setTimestamp(timestamp);
+
+            String dataType = metadata.get("dataType").getAsString();
+            switch (dataType) {
+                case "Boolean":
+                    message.setValue(value.getAsBoolean());
+                    message.setDataType(DataType.BOOLEAN);
+                    break;
+                case "Char":
+                    message.setValue(value.getAsString().getBytes(StandardCharsets.UTF_8));
+                    message.setDataType(DataType.BINARY);
+                    break;
+                case "Byte":
+                    message.setValue(value.getAsInt());
+                    message.setDataType(DataType.INTEGER);
+                    break;
+                default:
+                    logger.warn("unknown datatype of mqtt: " + dataType);
+            }
+            messages.add(message);
+        }
+        return messages;
+    }
+}