From e216a369b3c0b585bf961ca4d324e5e53a70618e Mon Sep 17 00:00:00 2001 From: pengys5 <8082209@qq.com> Date: Sat, 29 Jul 2017 23:01:22 +0800 Subject: [PATCH] Segment save to es success --- .../component/dao/NodeComponentEsDAO.java | 2 +- .../node/mapping/dao/NodeMappingEsDAO.java | 2 +- .../reference/dao/NodeReferenceEsDAO.java | 2 +- .../worker/segment/SegmentParse.java | 26 ++++++ .../segment/SegmentPersistenceWorker.java | 70 ++++++++++++++++ .../worker/segment/dao/ISegmentDAO.java | 12 +++ .../worker/segment/dao/SegmentEsDAO.java | 34 ++++++++ .../worker/segment/dao/SegmentH2DAO.java | 15 ++++ .../segment/define/SegmentDataDefine.java | 83 +++++++++++++++++++ .../segment/define/SegmentEsTableDefine.java | 30 +++++++ .../segment/define/SegmentH2TableDefine.java | 19 +++++ .../worker/segment/define/SegmentTable.java | 11 +++ .../resources/META-INF/defines/es_dao.define | 3 +- .../resources/META-INF/defines/h2_dao.define | 3 +- .../local_async_worker_provider.define | 2 + .../resources/META-INF/defines/storage.define | 5 +- .../core/storage/StorageInstaller.java | 1 + .../define/ElasticSearchStorageInstaller.java | 3 +- .../storage/h2/define/H2ColumnDefine.java | 2 +- .../stream/worker/impl/AggregationWorker.java | 2 +- .../worker/impl/data/AttributeType.java | 2 +- .../stream/worker/impl/data/Data.java | 20 ++++- .../stream/worker/impl/data/DataDefine.java | 9 +- .../stream/worker/impl/data/Operation.java | 2 + .../impl/data/operate/CoverOperation.java | 4 + .../impl/data/operate/NonOperation.java | 4 + .../src/main/proto/RemoteCommonService.proto | 10 ++- 27 files changed, 362 insertions(+), 16 deletions(-) create mode 100644 apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/SegmentPersistenceWorker.java create mode 100644 apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/dao/ISegmentDAO.java create mode 100644 apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/dao/SegmentEsDAO.java create mode 100644 apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/dao/SegmentH2DAO.java create mode 100644 apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/define/SegmentDataDefine.java create mode 100644 apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/define/SegmentEsTableDefine.java create mode 100644 apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/define/SegmentH2TableDefine.java create mode 100644 apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/define/SegmentTable.java diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/node/component/dao/NodeComponentEsDAO.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/node/component/dao/NodeComponentEsDAO.java index 97840799f..012b508e6 100644 --- a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/node/component/dao/NodeComponentEsDAO.java +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/node/component/dao/NodeComponentEsDAO.java @@ -21,7 +21,7 @@ public class NodeComponentEsDAO extends EsDAO implements INodeComponentDAO { source.put(NodeComponentTable.COLUMN_AGG, data.getDataString(1)); source.put(NodeComponentTable.COLUMN_TIME_BUCKET, data.getDataLong(0)); - IndexRequestBuilder builder = getClient().prepareIndex(NodeComponentTable.TABLE, id).setSource(); + IndexRequestBuilder builder = getClient().prepareIndex(NodeComponentTable.TABLE, id).setSource(source); indexRequestBuilders.add(builder); }); return indexRequestBuilders; diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/node/mapping/dao/NodeMappingEsDAO.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/node/mapping/dao/NodeMappingEsDAO.java index 7054ff861..4a36d0378 100644 --- a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/node/mapping/dao/NodeMappingEsDAO.java +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/node/mapping/dao/NodeMappingEsDAO.java @@ -21,7 +21,7 @@ public class NodeMappingEsDAO extends EsDAO implements INodeMappingDAO { source.put(NodeMappingTable.COLUMN_AGG, data.getDataString(1)); source.put(NodeMappingTable.COLUMN_TIME_BUCKET, data.getDataLong(0)); - IndexRequestBuilder builder = getClient().prepareIndex(NodeMappingTable.TABLE, id).setSource(); + IndexRequestBuilder builder = getClient().prepareIndex(NodeMappingTable.TABLE, id).setSource(source); indexRequestBuilders.add(builder); }); return indexRequestBuilders; diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/noderef/reference/dao/NodeReferenceEsDAO.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/noderef/reference/dao/NodeReferenceEsDAO.java index 114023418..d1334bec6 100644 --- a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/noderef/reference/dao/NodeReferenceEsDAO.java +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/noderef/reference/dao/NodeReferenceEsDAO.java @@ -21,7 +21,7 @@ public class NodeReferenceEsDAO extends EsDAO implements INodeReferenceDAO { source.put(NodeRefTable.COLUMN_AGG, data.getDataString(1)); source.put(NodeRefTable.COLUMN_TIME_BUCKET, data.getDataLong(0)); - IndexRequestBuilder builder = getClient().prepareIndex(NodeRefTable.TABLE, id).setSource(); + IndexRequestBuilder builder = getClient().prepareIndex(NodeRefTable.TABLE, id).setSource(source); indexRequestBuilders.add(builder); }); return indexRequestBuilders; diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/SegmentParse.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/SegmentParse.java index 60dba44d6..701109755 100644 --- a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/SegmentParse.java +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/SegmentParse.java @@ -5,7 +5,13 @@ import java.util.List; import org.skywalking.apm.collector.agentstream.worker.node.component.NodeComponentSpanListener; import org.skywalking.apm.collector.agentstream.worker.node.mapping.NodeMappingSpanListener; import org.skywalking.apm.collector.agentstream.worker.noderef.reference.NodeRefSpanListener; +import org.skywalking.apm.collector.agentstream.worker.segment.define.SegmentDataDefine; +import org.skywalking.apm.collector.core.framework.CollectorContextHelper; import org.skywalking.apm.collector.core.util.CollectionUtils; +import org.skywalking.apm.collector.stream.StreamModuleContext; +import org.skywalking.apm.collector.stream.StreamModuleGroupDefine; +import org.skywalking.apm.collector.stream.worker.WorkerInvokeException; +import org.skywalking.apm.collector.stream.worker.WorkerNotFoundException; import org.skywalking.apm.network.proto.SpanObject; import org.skywalking.apm.network.proto.SpanType; import org.skywalking.apm.network.proto.TraceSegmentObject; @@ -68,6 +74,26 @@ public class SegmentParse { } notifyListenerToBuild(); + + StringBuilder segmentId = new StringBuilder(); + segmentObject.getTraceSegmentId().getIdPartsList().forEach(part -> { + segmentId.append(part); + }); + buildSegment(segmentId.toString(), segmentObject.toByteArray()); + } + + public void buildSegment(String id, byte[] dataBinary) { + StreamModuleContext context = (StreamModuleContext)CollectorContextHelper.INSTANCE.getContext(StreamModuleGroupDefine.GROUP_NAME); + SegmentDataDefine.Segment segment = new SegmentDataDefine.Segment(); + segment.setId(id); + segment.setDataBinary(dataBinary); + + try { + logger.debug("send to segment persistence worker, id: {}, dataBinary length: {}", segment.getId(), dataBinary.length); + context.getClusterWorkerContext().lookup(SegmentPersistenceWorker.WorkerRole.INSTANCE).tell(segment.transform()); + } catch (WorkerInvokeException | WorkerNotFoundException e) { + logger.error(e.getMessage(), e); + } } private void notifyListenerToBuild() { diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/SegmentPersistenceWorker.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/SegmentPersistenceWorker.java new file mode 100644 index 000000000..7494c955d --- /dev/null +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/SegmentPersistenceWorker.java @@ -0,0 +1,70 @@ +package org.skywalking.apm.collector.agentstream.worker.segment; + +import java.util.List; +import java.util.Map; +import org.skywalking.apm.collector.agentstream.worker.segment.dao.ISegmentDAO; +import org.skywalking.apm.collector.agentstream.worker.segment.define.SegmentDataDefine; +import org.skywalking.apm.collector.storage.dao.DAOContainer; +import org.skywalking.apm.collector.stream.worker.AbstractLocalAsyncWorkerProvider; +import org.skywalking.apm.collector.stream.worker.ClusterWorkerContext; +import org.skywalking.apm.collector.stream.worker.ProviderNotFoundException; +import org.skywalking.apm.collector.stream.worker.Role; +import org.skywalking.apm.collector.stream.worker.impl.PersistenceWorker; +import org.skywalking.apm.collector.stream.worker.impl.data.Data; +import org.skywalking.apm.collector.stream.worker.impl.data.DataDefine; +import org.skywalking.apm.collector.stream.worker.selector.RollingSelector; +import org.skywalking.apm.collector.stream.worker.selector.WorkerSelector; + +/** + * @author pengys5 + */ +public class SegmentPersistenceWorker extends PersistenceWorker { + + public SegmentPersistenceWorker(Role role, ClusterWorkerContext clusterContext) { + super(role, clusterContext); + } + + @Override public void preStart() throws ProviderNotFoundException { + super.preStart(); + } + + @Override protected List prepareBatch(Map dataMap) { + ISegmentDAO dao = (ISegmentDAO)DAOContainer.INSTANCE.get(ISegmentDAO.class.getName()); + return dao.prepareBatch(dataMap); + } + + public static class Factory extends AbstractLocalAsyncWorkerProvider { + @Override + public Role role() { + return WorkerRole.INSTANCE; + } + + @Override + public SegmentPersistenceWorker workerInstance(ClusterWorkerContext clusterContext) { + return new SegmentPersistenceWorker(role(), clusterContext); + } + + @Override + public int queueSize() { + return 1024; + } + } + + public enum WorkerRole implements Role { + INSTANCE; + + @Override + public String roleName() { + return SegmentPersistenceWorker.class.getSimpleName(); + } + + @Override + public WorkerSelector workerSelector() { + return new RollingSelector(); + } + + @Override public DataDefine dataDefine() { + return new SegmentDataDefine(); + } + } +} diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/dao/ISegmentDAO.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/dao/ISegmentDAO.java new file mode 100644 index 000000000..9db87a6aa --- /dev/null +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/dao/ISegmentDAO.java @@ -0,0 +1,12 @@ +package org.skywalking.apm.collector.agentstream.worker.segment.dao; + +import java.util.List; +import java.util.Map; +import org.skywalking.apm.collector.stream.worker.impl.data.Data; + +/** + * @author pengys5 + */ +public interface ISegmentDAO { + List prepareBatch(Map dataMap); +} diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/dao/SegmentEsDAO.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/dao/SegmentEsDAO.java new file mode 100644 index 000000000..f51741963 --- /dev/null +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/dao/SegmentEsDAO.java @@ -0,0 +1,34 @@ +package org.skywalking.apm.collector.agentstream.worker.segment.dao; + +import java.util.ArrayList; +import java.util.Base64; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import org.elasticsearch.action.index.IndexRequestBuilder; +import org.skywalking.apm.collector.agentstream.worker.segment.define.SegmentTable; +import org.skywalking.apm.collector.storage.elasticsearch.dao.EsDAO; +import org.skywalking.apm.collector.stream.worker.impl.data.Data; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * @author pengys5 + */ +public class SegmentEsDAO extends EsDAO implements ISegmentDAO { + + private final Logger logger = LoggerFactory.getLogger(SegmentEsDAO.class); + + @Override public List prepareBatch(Map dataMap) { + List indexRequestBuilders = new ArrayList<>(); + dataMap.forEach((id, data) -> { + logger.debug("segment prepareBatch, id: {}", id); + Map source = new HashMap(); + source.put(SegmentTable.COLUMN_DATA_BINARY, new String(Base64.getEncoder().encode(data.getDataBytes(0)))); + logger.debug("segment source: {}", source.toString()); + IndexRequestBuilder builder = getClient().prepareIndex(SegmentTable.TABLE, id).setSource(source); + indexRequestBuilders.add(builder); + }); + return indexRequestBuilders; + } +} diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/dao/SegmentH2DAO.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/dao/SegmentH2DAO.java new file mode 100644 index 000000000..adf49c8fb --- /dev/null +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/dao/SegmentH2DAO.java @@ -0,0 +1,15 @@ +package org.skywalking.apm.collector.agentstream.worker.segment.dao; + +import java.util.List; +import java.util.Map; +import org.skywalking.apm.collector.storage.h2.dao.H2DAO; + +/** + * @author pengys5 + */ +public class SegmentH2DAO extends H2DAO implements ISegmentDAO { + + @Override public List prepareBatch(Map map) { + return null; + } +} diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/define/SegmentDataDefine.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/define/SegmentDataDefine.java new file mode 100644 index 000000000..f77e3e184 --- /dev/null +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/define/SegmentDataDefine.java @@ -0,0 +1,83 @@ +package org.skywalking.apm.collector.agentstream.worker.segment.define; + +import com.google.protobuf.ByteString; +import org.skywalking.apm.collector.remote.grpc.proto.RemoteData; +import org.skywalking.apm.collector.stream.worker.impl.data.Attribute; +import org.skywalking.apm.collector.stream.worker.impl.data.AttributeType; +import org.skywalking.apm.collector.stream.worker.impl.data.Data; +import org.skywalking.apm.collector.stream.worker.impl.data.DataDefine; +import org.skywalking.apm.collector.stream.worker.impl.data.TransformToData; +import org.skywalking.apm.collector.stream.worker.impl.data.operate.CoverOperation; +import org.skywalking.apm.collector.stream.worker.impl.data.operate.NonOperation; + +/** + * @author pengys5 + */ +public class SegmentDataDefine extends DataDefine { + + public static final int DEFINE_ID = 401; + + @Override public int defineId() { + return DEFINE_ID; + } + + @Override protected int initialCapacity() { + return 2; + } + + @Override protected void attributeDefine() { + addAttribute(0, new Attribute(SegmentTable.COLUMN_ID, AttributeType.STRING, new NonOperation())); + addAttribute(1, new Attribute(SegmentTable.COLUMN_DATA_BINARY, AttributeType.BYTE, new CoverOperation())); + } + + @Override public Object deserialize(RemoteData remoteData) { + String id = remoteData.getDataStrings(0); + byte[] dataBinary = remoteData.getDataBytes(0).toByteArray(); + return new Segment(id, dataBinary); + } + + @Override public RemoteData serialize(Object object) { + Segment segment = (Segment)object; + RemoteData.Builder builder = RemoteData.newBuilder(); + builder.addDataStrings(segment.getId()); + builder.addDataBytes(ByteString.copyFrom(segment.getDataBinary())); + return builder.build(); + } + + public static class Segment implements TransformToData { + private String id; + private byte[] dataBinary; + + public Segment(String id, byte[] dataBinary) { + this.id = id; + this.dataBinary = dataBinary; + } + + public Segment() { + } + + @Override public Data transform() { + SegmentDataDefine define = new SegmentDataDefine(); + Data data = define.build(id); + data.setDataString(0, this.id); + data.setDataBytes(0, this.dataBinary); + return data; + } + + public String getId() { + return id; + } + + public void setId(String id) { + this.id = id; + } + + public byte[] getDataBinary() { + return dataBinary; + } + + public void setDataBinary(byte[] dataBinary) { + this.dataBinary = dataBinary; + } + } +} diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/define/SegmentEsTableDefine.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/define/SegmentEsTableDefine.java new file mode 100644 index 000000000..8635e1252 --- /dev/null +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/define/SegmentEsTableDefine.java @@ -0,0 +1,30 @@ +package org.skywalking.apm.collector.agentstream.worker.segment.define; + +import org.skywalking.apm.collector.storage.elasticsearch.define.ElasticSearchColumnDefine; +import org.skywalking.apm.collector.storage.elasticsearch.define.ElasticSearchTableDefine; + +/** + * @author pengys5 + */ +public class SegmentEsTableDefine extends ElasticSearchTableDefine { + + public SegmentEsTableDefine() { + super(SegmentTable.TABLE); + } + + @Override public int refreshInterval() { + return 10; + } + + @Override public int numberOfShards() { + return 2; + } + + @Override public int numberOfReplicas() { + return 0; + } + + @Override public void initialize() { + addColumn(new ElasticSearchColumnDefine(SegmentTable.COLUMN_DATA_BINARY, ElasticSearchColumnDefine.Type.Binary.name())); + } +} diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/define/SegmentH2TableDefine.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/define/SegmentH2TableDefine.java new file mode 100644 index 000000000..2f494b8f5 --- /dev/null +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/define/SegmentH2TableDefine.java @@ -0,0 +1,19 @@ +package org.skywalking.apm.collector.agentstream.worker.segment.define; + +import org.skywalking.apm.collector.storage.h2.define.H2ColumnDefine; +import org.skywalking.apm.collector.storage.h2.define.H2TableDefine; + +/** + * @author pengys5 + */ +public class SegmentH2TableDefine extends H2TableDefine { + + public SegmentH2TableDefine() { + super(SegmentTable.TABLE); + } + + @Override public void initialize() { + addColumn(new H2ColumnDefine(SegmentTable.COLUMN_ID, H2ColumnDefine.Type.Varchar.name())); + addColumn(new H2ColumnDefine(SegmentTable.COLUMN_DATA_BINARY, H2ColumnDefine.Type.BINARY.name())); + } +} diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/define/SegmentTable.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/define/SegmentTable.java new file mode 100644 index 000000000..8f3cc3e45 --- /dev/null +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/define/SegmentTable.java @@ -0,0 +1,11 @@ +package org.skywalking.apm.collector.agentstream.worker.segment.define; + +import org.skywalking.apm.collector.agentstream.worker.CommonTable; + +/** + * @author pengys5 + */ +public class SegmentTable extends CommonTable { + public static final String TABLE = "segment"; + public static final String COLUMN_DATA_BINARY = "data_binary"; +} diff --git a/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/es_dao.define b/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/es_dao.define index 1c32ea41e..0eff3bcc1 100644 --- a/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/es_dao.define +++ b/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/es_dao.define @@ -3,4 +3,5 @@ org.skywalking.apm.collector.agentstream.worker.register.instance.dao.InstanceEs org.skywalking.apm.collector.agentstream.worker.register.servicename.dao.ServiceNameEsDAO org.skywalking.apm.collector.agentstream.worker.node.component.dao.NodeComponentEsDAO org.skywalking.apm.collector.agentstream.worker.node.mapping.dao.NodeMappingEsDAO -org.skywalking.apm.collector.agentstream.worker.noderef.reference.dao.NodeReferenceEsDAO \ No newline at end of file +org.skywalking.apm.collector.agentstream.worker.noderef.reference.dao.NodeReferenceEsDAO +org.skywalking.apm.collector.agentstream.worker.segment.dao.SegmentEsDAO \ No newline at end of file diff --git a/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/h2_dao.define b/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/h2_dao.define index a83ad8a26..0d5afd43f 100644 --- a/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/h2_dao.define +++ b/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/h2_dao.define @@ -3,4 +3,5 @@ org.skywalking.apm.collector.agentstream.worker.register.instance.dao.InstanceH2 org.skywalking.apm.collector.agentstream.worker.register.servicename.dao.ServiceNameH2DAO org.skywalking.apm.collector.agentstream.worker.node.component.dao.NodeComponentH2DAO org.skywalking.apm.collector.agentstream.worker.node.mapping.dao.NodeMappingH2DAO -org.skywalking.apm.collector.agentstream.worker.noderef.reference.dao.NodeReferenceH2DAO \ No newline at end of file +org.skywalking.apm.collector.agentstream.worker.noderef.reference.dao.NodeReferenceH2DAO +org.skywalking.apm.collector.agentstream.worker.segment.dao.SegmentH2DAO \ No newline at end of file diff --git a/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/local_async_worker_provider.define b/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/local_async_worker_provider.define index 208c1813c..389c3a19d 100644 --- a/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/local_async_worker_provider.define +++ b/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/local_async_worker_provider.define @@ -7,6 +7,8 @@ org.skywalking.apm.collector.agentstream.worker.node.mapping.NodeMappingPersiste org.skywalking.apm.collector.agentstream.worker.noderef.reference.NodeRefAggregationWorker$Factory org.skywalking.apm.collector.agentstream.worker.noderef.reference.NodeRefPersistenceWorker$Factory +org.skywalking.apm.collector.agentstream.worker.segment.SegmentPersistenceWorker$Factory + org.skywalking.apm.collector.agentstream.worker.register.application.ApplicationRegisterSerialWorker$Factory org.skywalking.apm.collector.agentstream.worker.register.instance.InstanceRegisterSerialWorker$Factory org.skywalking.apm.collector.agentstream.worker.register.servicename.ServiceNameRegisterSerialWorker$Factory \ No newline at end of file diff --git a/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/storage.define b/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/storage.define index 7e372faa8..2bb31fb38 100644 --- a/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/storage.define +++ b/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/storage.define @@ -14,4 +14,7 @@ org.skywalking.apm.collector.agentstream.worker.register.instance.InstanceEsTabl org.skywalking.apm.collector.agentstream.worker.register.instance.InstanceH2TableDefine org.skywalking.apm.collector.agentstream.worker.register.servicename.ServiceNameEsTableDefine -org.skywalking.apm.collector.agentstream.worker.register.servicename.ServiceNameH2TableDefine \ No newline at end of file +org.skywalking.apm.collector.agentstream.worker.register.servicename.ServiceNameH2TableDefine + +org.skywalking.apm.collector.agentstream.worker.segment.define.SegmentEsTableDefine +org.skywalking.apm.collector.agentstream.worker.segment.define.SegmentH2TableDefine \ No newline at end of file diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/storage/StorageInstaller.java b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/storage/StorageInstaller.java index c90bb3db2..0ae185794 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/storage/StorageInstaller.java +++ b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/storage/StorageInstaller.java @@ -22,6 +22,7 @@ public abstract class StorageInstaller { for (TableDefine tableDefine : tableDefines) { if (!isExists(client, tableDefine)) { logger.info("table: {} not exists", tableDefine.getName()); + tableDefine.initialize(); createTable(client, tableDefine); } } diff --git a/apm-collector/apm-collector-storage/src/main/java/org/skywalking/apm/collector/storage/elasticsearch/define/ElasticSearchStorageInstaller.java b/apm-collector/apm-collector-storage/src/main/java/org/skywalking/apm/collector/storage/elasticsearch/define/ElasticSearchStorageInstaller.java index 6cdd78dc5..25dd33d61 100644 --- a/apm-collector/apm-collector-storage/src/main/java/org/skywalking/apm/collector/storage/elasticsearch/define/ElasticSearchStorageInstaller.java +++ b/apm-collector/apm-collector-storage/src/main/java/org/skywalking/apm/collector/storage/elasticsearch/define/ElasticSearchStorageInstaller.java @@ -70,13 +70,14 @@ public class ElasticSearchStorageInstaller extends StorageInstaller { ElasticSearchColumnDefine elasticSearchColumnDefine = (ElasticSearchColumnDefine)columnDefine; mappingBuilder .startObject(elasticSearchColumnDefine.getName()) - .field("type", elasticSearchColumnDefine.getType()) + .field("type", elasticSearchColumnDefine.getType().toLowerCase()) .endObject(); } mappingBuilder .endObject() .endObject(); + logger.debug("create elasticsearch index: {}", mappingBuilder.string()); return mappingBuilder; } diff --git a/apm-collector/apm-collector-storage/src/main/java/org/skywalking/apm/collector/storage/h2/define/H2ColumnDefine.java b/apm-collector/apm-collector-storage/src/main/java/org/skywalking/apm/collector/storage/h2/define/H2ColumnDefine.java index fa7760a0d..ca7cf49f1 100644 --- a/apm-collector/apm-collector-storage/src/main/java/org/skywalking/apm/collector/storage/h2/define/H2ColumnDefine.java +++ b/apm-collector/apm-collector-storage/src/main/java/org/skywalking/apm/collector/storage/h2/define/H2ColumnDefine.java @@ -12,6 +12,6 @@ public class H2ColumnDefine extends ColumnDefine { } public enum Type { - Boolean, Varchar, Int, Bigint, Date + Boolean, Varchar, Int, Bigint, BINARY } } diff --git a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/AggregationWorker.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/AggregationWorker.java index 6d9dff63f..321b892b0 100644 --- a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/AggregationWorker.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/AggregationWorker.java @@ -71,7 +71,7 @@ public abstract class AggregationWorker extends AbstractLocalAsyncWorker { Data data = (Data)message; dataCache.hold(); if (dataCache.containsKey(data.id())) { - getClusterContext().getDataDefine(data.getDefineId()).mergeData(data, dataCache.get(data.id())); + getRole().dataDefine().mergeData(data, dataCache.get(data.id())); } else { dataCache.put(data.id(), data); } diff --git a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/AttributeType.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/AttributeType.java index 3b2856bd9..e2b298b32 100644 --- a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/AttributeType.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/AttributeType.java @@ -4,5 +4,5 @@ package org.skywalking.apm.collector.stream.worker.impl.data; * @author pengys5 */ public enum AttributeType { - STRING, LONG, FLOAT, INTEGER + STRING, LONG, FLOAT, INTEGER, BYTE } diff --git a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/Data.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/Data.java index 42ac82c28..9cca042e1 100644 --- a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/Data.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/Data.java @@ -1,5 +1,6 @@ package org.skywalking.apm.collector.stream.worker.impl.data; +import com.google.protobuf.ByteString; import org.skywalking.apm.collector.remote.grpc.proto.RemoteData; import org.skywalking.apm.collector.stream.worker.selector.AbstractHashMessage; @@ -12,22 +13,27 @@ public class Data extends AbstractHashMessage { private final int longCapacity; private final int floatCapacity; private final int integerCapacity; + private final int byteCapacity; private String[] dataStrings; private Long[] dataLongs; private Float[] dataFloats; private Integer[] dataIntegers; + private byte[][] dataBytes; - public Data(String id, int defineId, int stringCapacity, int longCapacity, int floatCapacity, int integerCapacity) { + public Data(String id, int defineId, int stringCapacity, int longCapacity, int floatCapacity, int integerCapacity, + int byteCapacity) { super(id); this.defineId = defineId; this.dataStrings = new String[stringCapacity]; this.dataLongs = new Long[longCapacity]; this.dataFloats = new Float[floatCapacity]; this.dataIntegers = new Integer[integerCapacity]; + this.dataBytes = new byte[byteCapacity][]; this.stringCapacity = stringCapacity; this.longCapacity = longCapacity; this.floatCapacity = floatCapacity; this.integerCapacity = integerCapacity; + this.byteCapacity = byteCapacity; } public void setDataString(int position, String value) { @@ -46,6 +52,10 @@ public class Data extends AbstractHashMessage { dataIntegers[position] = value; } + public void setDataBytes(int position, byte[] dataBytes) { + this.dataBytes[position] = dataBytes; + } + public String getDataString(int position) { return dataStrings[position]; } @@ -62,6 +72,10 @@ public class Data extends AbstractHashMessage { return dataIntegers[position]; } + public byte[] getDataBytes(int position) { + return dataBytes[position]; + } + public String id() { return dataStrings[0]; } @@ -76,6 +90,7 @@ public class Data extends AbstractHashMessage { builder.setFloatCapacity(floatCapacity); builder.setStringCapacity(stringCapacity); builder.setLongCapacity(longCapacity); + builder.setByteCapacity(byteCapacity); for (int i = 0; i < dataStrings.length; i++) { builder.setDataStrings(i, dataStrings[i]); @@ -89,6 +104,9 @@ public class Data extends AbstractHashMessage { for (int i = 0; i < dataLongs.length; i++) { builder.setDataLongs(i, dataLongs[i]); } + for (int i = 0; i < dataBytes.length; i++) { + builder.setDataBytes(i, ByteString.copyFrom(dataBytes[i])); + } return builder.build(); } } diff --git a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/DataDefine.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/DataDefine.java index 0c9174d6b..95af25ade 100644 --- a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/DataDefine.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/DataDefine.java @@ -11,6 +11,7 @@ public abstract class DataDefine { private int longCapacity; private int floatCapacity; private int integerCapacity; + private int byteCapacity; public DataDefine() { initial(); @@ -28,6 +29,8 @@ public abstract class DataDefine { floatCapacity++; } else if (AttributeType.INTEGER.equals(attribute.getType())) { integerCapacity++; + } else if (AttributeType.BYTE.equals(attribute.getType())) { + byteCapacity++; } } } @@ -43,7 +46,7 @@ public abstract class DataDefine { protected abstract void attributeDefine(); public final Data build(String id) { - return new Data(id, defineId(), stringCapacity, longCapacity, floatCapacity, integerCapacity); + return new Data(id, defineId(), stringCapacity, longCapacity, floatCapacity, integerCapacity, byteCapacity); } public void mergeData(Data newData, Data oldData) { @@ -51,6 +54,7 @@ public abstract class DataDefine { int longPosition = 0; int floatPosition = 0; int integerPosition = 0; + int bytePosition = 0; for (int i = 0; i < initialCapacity(); i++) { Attribute attribute = attributes[i]; if (AttributeType.STRING.equals(attribute.getType())) { @@ -65,6 +69,9 @@ public abstract class DataDefine { } else if (AttributeType.FLOAT.equals(attribute.getType())) { attribute.getOperation().operate(newData.getDataInteger(integerPosition), oldData.getDataInteger(integerPosition)); integerPosition++; + } else if (AttributeType.BYTE.equals(attribute.getType())) { + attribute.getOperation().operate(newData.getDataBytes(bytePosition), oldData.getDataBytes(integerPosition)); + bytePosition++; } } } diff --git a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/Operation.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/Operation.java index fea3f9f87..395617fbe 100644 --- a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/Operation.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/Operation.java @@ -11,4 +11,6 @@ public interface Operation { Float operate(Float newValue, Float oldValue); Integer operate(Integer newValue, Integer oldValue); + + byte[] operate(byte[] newValue, byte[] oldValue); } diff --git a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/operate/CoverOperation.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/operate/CoverOperation.java index 99c4eb31b..ffc9a1b1a 100644 --- a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/operate/CoverOperation.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/operate/CoverOperation.java @@ -21,4 +21,8 @@ public class CoverOperation implements Operation { @Override public Integer operate(Integer newValue, Integer oldValue) { return newValue; } + + @Override public byte[] operate(byte[] newValue, byte[] oldValue) { + return newValue; + } } diff --git a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/operate/NonOperation.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/operate/NonOperation.java index 7bc09ba17..33ca08b01 100644 --- a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/operate/NonOperation.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/operate/NonOperation.java @@ -21,4 +21,8 @@ public class NonOperation implements Operation { @Override public Integer operate(Integer newValue, Integer oldValue) { return oldValue; } + + @Override public byte[] operate(byte[] newValue, byte[] oldValue) { + return oldValue; + } } diff --git a/apm-collector/apm-collector-stream/src/main/proto/RemoteCommonService.proto b/apm-collector/apm-collector-stream/src/main/proto/RemoteCommonService.proto index 4160d7c64..5b3122af3 100644 --- a/apm-collector/apm-collector-stream/src/main/proto/RemoteCommonService.proto +++ b/apm-collector/apm-collector-stream/src/main/proto/RemoteCommonService.proto @@ -18,10 +18,12 @@ message RemoteData { int32 longCapacity = 2; int32 floatCapacity = 3; int32 integerCapacity = 4; - repeated string dataStrings = 5; - repeated int64 dataLongs = 6; - repeated float dataFloats = 7; - repeated int32 dataIntegers = 8; + int32 byteCapacity = 5; + repeated string dataStrings = 6; + repeated int64 dataLongs = 7; + repeated float dataFloats = 8; + repeated int32 dataIntegers = 9; + repeated bytes dataBytes = 10; } message Empty {