diff --git a/oap-server/generate-tool-grammar/src/main/antlr4/org/apache/skywalking/oal/tool/grammar/OALLexer.g4 b/oap-server/generate-tool-grammar/src/main/antlr4/org/apache/skywalking/oal/tool/grammar/OALLexer.g4 index b9cb869ca..8ae21e244 100644 --- a/oap-server/generate-tool-grammar/src/main/antlr4/org/apache/skywalking/oal/tool/grammar/OALLexer.g4 +++ b/oap-server/generate-tool-grammar/src/main/antlr4/org/apache/skywalking/oal/tool/grammar/OALLexer.g4 @@ -51,6 +51,7 @@ SRC_ENDPOINT_RELATION_SERVER_SIDE: 'endpoint_relation_server_side'; SRC_SERVICE_RELATION_SERVER_SIDE: 'service_relation_server_side'; SRC_SERVICE_RELATION_CLIENT_SIDE: 'service_relation_client_side'; SRC_ALARM_RECORD: 'alarm_record'; +SRC_HTTP_ACCESS_LOG: 'http_access_log'; SRC_ZIPKIN_SPAN: 'zipkin_span'; SRC_JAEGER_SPAN: 'jaeger_span'; diff --git a/oap-server/generate-tool-grammar/src/main/antlr4/org/apache/skywalking/oal/tool/grammar/OALParser.g4 b/oap-server/generate-tool-grammar/src/main/antlr4/org/apache/skywalking/oal/tool/grammar/OALParser.g4 index 2622156d3..f407c02de 100644 --- a/oap-server/generate-tool-grammar/src/main/antlr4/org/apache/skywalking/oal/tool/grammar/OALParser.g4 +++ b/oap-server/generate-tool-grammar/src/main/antlr4/org/apache/skywalking/oal/tool/grammar/OALParser.g4 @@ -60,7 +60,7 @@ source disableSource : SRC_SEGMENT | SRC_TOP_N_DB_STATEMENT | SRC_ENDPOINT_RELATION_SERVER_SIDE | SRC_SERVICE_RELATION_SERVER_SIDE | - SRC_SERVICE_RELATION_CLIENT_SIDE | SRC_ALARM_RECORD + SRC_SERVICE_RELATION_CLIENT_SIDE | SRC_ALARM_RECORD | SRC_HTTP_ACCESS_LOG ; sourceAttribute diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModule.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModule.java index c53a48c5b..27855709b 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModule.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModule.java @@ -66,6 +66,7 @@ public class CoreModule extends ModuleDefine { classes.add(TopologyQueryService.class); classes.add(MetricQueryService.class); classes.add(TraceQueryService.class); + classes.add(LogQueryService.class); classes.add(MetadataQueryService.class); classes.add(AggregationQueryService.class); classes.add(AlarmQueryService.class); diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java index 0cfe991a3..9b677ec4b 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java @@ -145,6 +145,7 @@ public class CoreModuleProvider extends ModuleProvider { this.registerServiceImplementation(TopologyQueryService.class, new TopologyQueryService(getManager())); this.registerServiceImplementation(MetricQueryService.class, new MetricQueryService(getManager())); this.registerServiceImplementation(TraceQueryService.class, new TraceQueryService(getManager())); + this.registerServiceImplementation(LogQueryService.class, new LogQueryService(getManager())); this.registerServiceImplementation(MetadataQueryService.class, new MetadataQueryService(getManager())); this.registerServiceImplementation(AggregationQueryService.class, new AggregationQueryService(getManager())); this.registerServiceImplementation(AlarmQueryService.class, new AlarmQueryService(getManager())); diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/log/AbstractLogRecord.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/log/AbstractLogRecord.java new file mode 100644 index 000000000..799aeef77 --- /dev/null +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/log/AbstractLogRecord.java @@ -0,0 +1,88 @@ +/* + * 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.skywalking.oap.server.core.analysis.manual.log; + +import java.util.*; +import lombok.*; +import org.apache.skywalking.oap.server.core.UnexpectedException; +import org.apache.skywalking.oap.server.core.analysis.record.Record; +import org.apache.skywalking.oap.server.core.query.entity.ContentType; +import org.apache.skywalking.oap.server.core.storage.StorageBuilder; +import org.apache.skywalking.oap.server.core.storage.annotation.Column; + +/** + * @author wusheng + */ + +public abstract class AbstractLogRecord extends Record { + + public static final String SERVICE_ID = "service_id"; + public static final String SERVICE_INSTANCE_ID = "service_instance_id"; + public static final String ENDPOINT_ID = "endpoint_id"; + public static final String TRACE_ID = "trace_id"; + public static final String IS_ERROR = "is_error"; + public static final String STATUS_CODE = "status_code"; + public static final String CONTENT_TYPE = "content_type"; + public static final String CONTENT = "content"; + public static final String TIMESTAMP = "timestamp"; + + @Setter @Getter @Column(columnName = SERVICE_ID) private int serviceId; + @Setter @Getter @Column(columnName = SERVICE_INSTANCE_ID) private int serviceInstanceId; + @Setter @Getter @Column(columnName = ENDPOINT_ID) private int endpointId; + @Setter @Getter @Column(columnName = TRACE_ID) private String traceId; + @Setter @Getter @Column(columnName = IS_ERROR) private int isError; + @Setter @Getter @Column(columnName = STATUS_CODE) private String statusCode; + @Setter @Getter @Column(columnName = CONTENT_TYPE) private int contentType = ContentType.NONE.value(); + @Setter @Getter @Column(columnName = CONTENT) private String content; + @Setter @Getter @Column(columnName = TIMESTAMP) private long timestamp; + + @Override public String id() { + throw new UnexpectedException("AbstractLogRecord doesn't provide id()"); + } + + public static abstract class Builder implements StorageBuilder { + protected void map2Data(T record, Map dbMap) { + record.setServiceId(((Number)dbMap.get(SERVICE_ID)).intValue()); + record.setServiceInstanceId(((Number)dbMap.get(SERVICE_INSTANCE_ID)).intValue()); + record.setEndpointId(((Number)dbMap.get(ENDPOINT_ID)).intValue()); + record.setIsError(((Number)dbMap.get(IS_ERROR)).intValue()); + record.setTraceId((String)dbMap.get(TRACE_ID)); + record.setStatusCode((String)dbMap.get(STATUS_CODE)); + record.setContentType(((Number)dbMap.get(CONTENT_TYPE)).intValue()); + record.setContent((String)dbMap.get(CONTENT)); + record.setTimestamp(((Number)dbMap.get(TIMESTAMP)).longValue()); + record.setTimeBucket(((Number)dbMap.get(TIME_BUCKET)).longValue()); + } + + @Override public Map data2Map(AbstractLogRecord record) { + Map map = new HashMap<>(); + map.put(SERVICE_ID, record.getServiceId()); + map.put(SERVICE_INSTANCE_ID, record.getServiceInstanceId()); + map.put(ENDPOINT_ID, record.getEndpointId()); + map.put(TRACE_ID, record.getTraceId()); + map.put(IS_ERROR, record.getIsError()); + map.put(STATUS_CODE, record.getStatusCode()); + map.put(TIME_BUCKET, record.getTimeBucket()); + map.put(CONTENT_TYPE, record.getContentType()); + map.put(CONTENT, record.getContent()); + map.put(TIMESTAMP, record.getTimestamp()); + return map; + } + } +} diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/log/HTTPAccessLogDispatcher.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/log/HTTPAccessLogDispatcher.java new file mode 100644 index 000000000..6a73a1229 --- /dev/null +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/log/HTTPAccessLogDispatcher.java @@ -0,0 +1,44 @@ +/* + * 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.skywalking.oap.server.core.analysis.manual.log; + +import org.apache.skywalking.oap.server.core.analysis.SourceDispatcher; +import org.apache.skywalking.oap.server.core.analysis.worker.RecordProcess; +import org.apache.skywalking.oap.server.core.source.HTTPAccessLog; + +/** + * @author wusheng + */ +public class HTTPAccessLogDispatcher implements SourceDispatcher { + @Override public void dispatch(HTTPAccessLog source) { + HTTPAccessLogRecord record = new HTTPAccessLogRecord(); + record.setTimestamp(source.getTimestamp()); + record.setTimeBucket(source.getTimeBucket()); + record.setServiceId(source.getServiceId()); + record.setServiceInstanceId(source.getServiceInstanceId()); + record.setEndpointId(source.getEndpointId()); + record.setTraceId(source.getTraceId()); + record.setIsError(source.getIsError()); + record.setStatusCode(source.getStatusCode()); + record.setContentType(source.getContentType().value()); + record.setContent(source.getContent()); + + RecordProcess.INSTANCE.in(record); + } +} diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/log/HTTPAccessLogRecord.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/log/HTTPAccessLogRecord.java new file mode 100644 index 000000000..af29461a1 --- /dev/null +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/log/HTTPAccessLogRecord.java @@ -0,0 +1,38 @@ +/* + * 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.skywalking.oap.server.core.analysis.manual.log; + +import java.util.Map; +import org.apache.skywalking.oap.server.core.source.DefaultScopeDefine; +import org.apache.skywalking.oap.server.core.storage.annotation.StorageEntity; + +import static org.apache.skywalking.oap.server.core.analysis.manual.log.HTTPAccessLogRecord.INDEX_NAME; + +@StorageEntity(name = INDEX_NAME, builder = HTTPAccessLogRecord.Builder.class, sourceScopeId = DefaultScopeDefine.HTTP_ACCESS_LOG) +public class HTTPAccessLogRecord extends AbstractLogRecord { + public static final String INDEX_NAME = "http_access_log"; + + public static class Builder extends AbstractLogRecord.Builder { + @Override public HTTPAccessLogRecord map2Data(Map dbMap) { + HTTPAccessLogRecord record = new HTTPAccessLogRecord(); + super.map2Data(record, dbMap); + return record; + } + } +} diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/segment/SegmentRecord.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/segment/SegmentRecord.java index ae1b56c9d..6c386d2e6 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/segment/SegmentRecord.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/segment/SegmentRecord.java @@ -50,18 +50,18 @@ public class SegmentRecord extends Record { public static final String DATA_BINARY = "data_binary"; public static final String VERSION = "version"; - @Setter @Getter @Column(columnName = SEGMENT_ID) @IDColumn private String segmentId; - @Setter @Getter @Column(columnName = TRACE_ID) @IDColumn private String traceId; - @Setter @Getter @Column(columnName = SERVICE_ID) @IDColumn private int serviceId; - @Setter @Getter @Column(columnName = SERVICE_INSTANCE_ID) @IDColumn private int serviceInstanceId; - @Setter @Getter @Column(columnName = ENDPOINT_NAME, matchQuery = true) @IDColumn private String endpointName; - @Setter @Getter @Column(columnName = ENDPOINT_ID) @IDColumn private int endpointId; - @Setter @Getter @Column(columnName = START_TIME) @IDColumn private long startTime; - @Setter @Getter @Column(columnName = END_TIME) @IDColumn private long endTime; - @Setter @Getter @Column(columnName = LATENCY) @IDColumn private int latency; - @Setter @Getter @Column(columnName = IS_ERROR) @IDColumn private int isError; - @Setter @Getter @Column(columnName = DATA_BINARY) @IDColumn private byte[] dataBinary; - @Setter @Getter @Column(columnName = VERSION) @IDColumn private int version; + @Setter @Getter @Column(columnName = SEGMENT_ID) private String segmentId; + @Setter @Getter @Column(columnName = TRACE_ID) private String traceId; + @Setter @Getter @Column(columnName = SERVICE_ID) private int serviceId; + @Setter @Getter @Column(columnName = SERVICE_INSTANCE_ID) private int serviceInstanceId; + @Setter @Getter @Column(columnName = ENDPOINT_NAME, matchQuery = true) private String endpointName; + @Setter @Getter @Column(columnName = ENDPOINT_ID) private int endpointId; + @Setter @Getter @Column(columnName = START_TIME) private long startTime; + @Setter @Getter @Column(columnName = END_TIME) private long endTime; + @Setter @Getter @Column(columnName = LATENCY) private int latency; + @Setter @Getter @Column(columnName = IS_ERROR) private int isError; + @Setter @Getter @Column(columnName = DATA_BINARY) private byte[] dataBinary; + @Setter @Getter @Column(columnName = VERSION) private int version; @Override public String id() { return segmentId; diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/LogQueryService.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/LogQueryService.java new file mode 100644 index 000000000..9a54a9562 --- /dev/null +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/LogQueryService.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 org.apache.skywalking.oap.server.core.query; + +import java.io.IOException; +import org.apache.skywalking.oap.server.core.*; +import org.apache.skywalking.oap.server.core.cache.*; +import org.apache.skywalking.oap.server.core.query.entity.*; +import org.apache.skywalking.oap.server.core.storage.StorageModule; +import org.apache.skywalking.oap.server.core.storage.query.ILogQueryDAO; +import org.apache.skywalking.oap.server.library.module.Service; +import org.apache.skywalking.oap.server.library.module.*; + +/** + * @author wusheng + */ +public class LogQueryService implements Service { + private final ModuleManager moduleManager; + private ILogQueryDAO logQueryDAO; + private ServiceInventoryCache serviceInventoryCache; + private ServiceInstanceInventoryCache serviceInstanceInventoryCache; + private EndpointInventoryCache endpointInventoryCache; + + public LogQueryService(ModuleManager moduleManager) { + this.moduleManager = moduleManager; + } + + private ILogQueryDAO getLogQueryDAO() { + if (logQueryDAO == null) { + this.logQueryDAO = moduleManager.find(StorageModule.NAME).provider().getService(ILogQueryDAO.class); + } + return logQueryDAO; + } + + private ServiceInventoryCache getServiceInventoryCache() { + if (serviceInventoryCache == null) { + this.serviceInventoryCache = moduleManager.find(CoreModule.NAME).provider().getService(ServiceInventoryCache.class); + } + return serviceInventoryCache; + } + + private ServiceInstanceInventoryCache getServiceInstanceInventoryCache() { + if (serviceInstanceInventoryCache == null) { + this.serviceInstanceInventoryCache = moduleManager.find(CoreModule.NAME).provider().getService(ServiceInstanceInventoryCache.class); + } + return serviceInstanceInventoryCache; + } + + private EndpointInventoryCache getEndpointInventoryCache() { + if (endpointInventoryCache == null) { + this.endpointInventoryCache = moduleManager.find(CoreModule.NAME).provider().getService(EndpointInventoryCache.class); + } + return endpointInventoryCache; + } + + public Logs queryLogs(final String metricName, int serviceId, int serviceInstanceId, int endpointId, + String traceId, LogState state, String stateCode, Pagination paging, final long startTB, + final long endTB) throws IOException { + PaginationUtils.Page page = PaginationUtils.INSTANCE.exchange(paging); + + Logs logs = getLogQueryDAO().queryLogs(metricName, serviceId, serviceInstanceId, endpointId, + traceId, state, stateCode, paging, page.getFrom(), page.getLimit(), startTB, endTB); + logs.getLogs().forEach(log -> { + if (log.getServiceId() != Const.NONE) { + log.setServiceName(getServiceInventoryCache().get(log.getServiceId()).getName()); + } + if (log.getServiceInstanceId() != Const.NONE) { + log.setServiceInstanceName(getServiceInstanceInventoryCache().get(log.getServiceInstanceId()).getName()); + } + if (log.getEndpointId() != Const.NONE) { + log.setEndpointName(getEndpointInventoryCache().get(log.getEndpointId()).getName()); + } + }); + return logs; + } +} diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/entity/ContentType.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/entity/ContentType.java new file mode 100644 index 000000000..68db82b24 --- /dev/null +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/entity/ContentType.java @@ -0,0 +1,51 @@ +/* + * 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.skywalking.oap.server.core.query.entity; + +import org.apache.skywalking.oap.server.core.UnexpectedException; + +/** + * @author wusheng + */ +public enum ContentType { + TEXT(1), JSON(2), NONE(0); + + private int value; + + ContentType(int value) { + this.value = value; + } + + public int value() { + return value; + } + + public static ContentType instanceOf(int value) { + switch (value) { + case 1: + return TEXT; + case 2: + return JSON; + case 0: + return NONE; + default: + throw new UnexpectedException("unexpected value=" + value); + } + } +} diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/entity/Log.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/entity/Log.java new file mode 100644 index 000000000..99cb3035a --- /dev/null +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/entity/Log.java @@ -0,0 +1,41 @@ +/* + * 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.skywalking.oap.server.core.query.entity; + +import lombok.*; + +/** + * @author wusheng + */ +@Setter +@Getter +public class Log { + private String serviceName; + private int serviceId; + private String serviceInstanceName; + private int serviceInstanceId; + private String endpointName; + private int endpointId; + private String traceId; + private String timestamp; + private boolean isError; + private String statusCode; + private ContentType contentType = ContentType.NONE; + private String content; +} diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/entity/LogState.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/entity/LogState.java new file mode 100644 index 000000000..dd262c92c --- /dev/null +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/entity/LogState.java @@ -0,0 +1,26 @@ +/* + * 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.skywalking.oap.server.core.query.entity; + +/** + * @author wusheng + */ +public enum LogState { + ALL, SUCCESS, ERROR +} diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/entity/Logs.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/entity/Logs.java new file mode 100644 index 000000000..738657641 --- /dev/null +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/entity/Logs.java @@ -0,0 +1,36 @@ +/* + * 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.skywalking.oap.server.core.query.entity; + +import java.util.*; +import lombok.*; + +/** + * @author wusheng + */ +@Setter +@Getter +public class Logs { + private final List logs; + private int total; + + public Logs() { + this.logs = new ArrayList<>(); + } +} diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/AbstractLog.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/AbstractLog.java new file mode 100644 index 000000000..a31efbf0a --- /dev/null +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/AbstractLog.java @@ -0,0 +1,42 @@ +/* + * 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.skywalking.oap.server.core.source; + +import lombok.*; +import org.apache.skywalking.oap.server.core.UnexpectedException; +import org.apache.skywalking.oap.server.core.query.entity.ContentType; + +@Setter +@Getter +public abstract class AbstractLog extends Source { + private long timeBucket; + private long timestamp; + private int serviceId; + private int serviceInstanceId; + private int endpointId; + private String traceId; + private int isError; + private String statusCode; + private ContentType contentType = ContentType.NONE; + private String content; + + @Override public String getEntityId() { + throw new UnexpectedException("getEntityId is not supported in AbstractLog source"); + } +} diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/DefaultScopeDefine.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/DefaultScopeDefine.java index b9c323915..4508868a1 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/DefaultScopeDefine.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/DefaultScopeDefine.java @@ -61,6 +61,7 @@ public class DefaultScopeDefine { public static final int ENVOY_INSTANCE_METRIC = 22; public static final int ZIPKIN_SPAN = 23; public static final int JAEGER_SPAN = 24; + public static final int HTTP_ACCESS_LOG = 25; /** * Catalog of scope, the indicator processor could use this to group all generated indicators by oal tool. diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/HTTPAccessLog.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/HTTPAccessLog.java new file mode 100644 index 000000000..3d163b4e5 --- /dev/null +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/HTTPAccessLog.java @@ -0,0 +1,31 @@ +/* + * 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.skywalking.oap.server.core.source; + +import static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.HTTP_ACCESS_LOG; + +/** + * @author wusheng + */ +@ScopeDeclaration(id = HTTP_ACCESS_LOG, name = "HTTPAccessLog") +public class HTTPAccessLog extends AbstractLog { + @Override public int scope() { + return HTTP_ACCESS_LOG; + } +} diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/StorageModule.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/StorageModule.java index 60729125a..eef5c124a 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/StorageModule.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/StorageModule.java @@ -40,6 +40,6 @@ public class StorageModule extends ModuleDefine { IServiceInventoryCacheDAO.class, IServiceInstanceInventoryCacheDAO.class, IEndpointInventoryCacheDAO.class, INetworkAddressInventoryCacheDAO.class, ITopologyQueryDAO.class, IMetricQueryDAO.class, ITraceQueryDAO.class, IMetadataQueryDAO.class, IAggregationQueryDAO.class, IAlarmQueryDAO.class, - ITopNRecordsQueryDAO.class}; + ITopNRecordsQueryDAO.class, ILogQueryDAO.class}; } } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/ILogQueryDAO.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/ILogQueryDAO.java new file mode 100644 index 000000000..ca8ed6a52 --- /dev/null +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/ILogQueryDAO.java @@ -0,0 +1,31 @@ +/* + * 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.skywalking.oap.server.core.storage.query; + +import java.io.IOException; +import org.apache.skywalking.oap.server.core.query.entity.*; +import org.apache.skywalking.oap.server.library.module.Service; + +/** + * @author wusheng + */ +public interface ILogQueryDAO extends Service { + Logs queryLogs(final String metricName, int serviceId, int serviceInstanceId, int endpointId, + String traceId, LogState state, String stateCode, Pagination paging, int from, int limit, final long startTB, final long endTB) throws IOException; +} diff --git a/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/GraphQLQueryProvider.java b/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/GraphQLQueryProvider.java index d940141d0..fcd3b245a 100644 --- a/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/GraphQLQueryProvider.java +++ b/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/GraphQLQueryProvider.java @@ -68,6 +68,8 @@ public class GraphQLQueryProvider extends ModuleProvider { .resolvers(new AlarmQuery(getManager())) .file("query-protocol/top-n-records.graphqls") .resolvers(new TopNRecordsQuery(getManager())) + .file("query-protocol/log.graphqls") + .resolvers(new LogQuery(getManager())) .build() .makeExecutableSchema(); this.graphQL = GraphQL.newGraphQL(schema).build(); diff --git a/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/LogQuery.java b/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/LogQuery.java new file mode 100644 index 000000000..cd48a3dce --- /dev/null +++ b/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/LogQuery.java @@ -0,0 +1,60 @@ +/* + * 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.skywalking.oap.query.graphql.resolver; + +import com.coxautodev.graphql.tools.GraphQLQueryResolver; +import java.io.IOException; +import org.apache.skywalking.oap.query.graphql.type.LogQueryCondition; +import org.apache.skywalking.oap.server.core.CoreModule; +import org.apache.skywalking.oap.server.core.query.*; +import org.apache.skywalking.oap.server.core.query.entity.Logs; +import org.apache.skywalking.oap.server.library.module.ModuleManager; + +import static java.util.Objects.nonNull; + +/** + * @author wusheng + */ +public class LogQuery implements GraphQLQueryResolver { + private final ModuleManager moduleManager; + private LogQueryService logQueryService; + + public LogQuery(ModuleManager moduleManager) { + this.moduleManager = moduleManager; + } + + private LogQueryService getQueryService() { + if (logQueryService == null) { + this.logQueryService = moduleManager.find(CoreModule.NAME).provider().getService(LogQueryService.class); + } + return logQueryService; + } + + public Logs queryLogs(LogQueryCondition condition) throws IOException { + long startSecondTB = 0; + long endSecondTB = 0; + if (nonNull(condition.getQueryDuration())) { + startSecondTB = DurationUtils.INSTANCE.startTimeDurationToSecondTimeBucket(condition.getQueryDuration().getStep(), condition.getQueryDuration().getStart()); + endSecondTB = DurationUtils.INSTANCE.endTimeDurationToSecondTimeBucket(condition.getQueryDuration().getStep(), condition.getQueryDuration().getEnd()); + } + + return getQueryService().queryLogs(condition.getMetricName(), condition.getServiceId(), condition.getServiceInstanceId(), condition.getEndpointId(), + condition.getTraceId(), condition.getState(), condition.getStateCode(), condition.getPaging(), startSecondTB, endSecondTB); + } +} diff --git a/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/type/LogQueryCondition.java b/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/type/LogQueryCondition.java new file mode 100644 index 000000000..3dc1130d1 --- /dev/null +++ b/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/type/LogQueryCondition.java @@ -0,0 +1,39 @@ +/* + * 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.skywalking.oap.query.graphql.type; + +import lombok.*; +import org.apache.skywalking.oap.server.core.query.entity.*; + +/** + * @author wusheng + */ +@Getter +@Setter +public class LogQueryCondition { + private String metricName; + private int serviceId; + private int serviceInstanceId; + private int endpointId; + private String traceId; + private LogState state; + private String stateCode; + private Duration queryDuration; + private Pagination paging; +} diff --git a/oap-server/server-query-plugin/query-graphql-plugin/src/main/resources/query-protocol b/oap-server/server-query-plugin/query-graphql-plugin/src/main/resources/query-protocol index 6f11e3b82..a005dd102 160000 --- a/oap-server/server-query-plugin/query-graphql-plugin/src/main/resources/query-protocol +++ b/oap-server/server-query-plugin/query-graphql-plugin/src/main/resources/query-protocol @@ -1 +1 @@ -Subproject commit 6f11e3b829bba4d3532477e968291cf657f0ac0b +Subproject commit a005dd1024c2b7e68fbcab5795824634cb81cde8 diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchProvider.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchProvider.java index ed3bc4baf..bff66df07 100644 --- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchProvider.java +++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchProvider.java @@ -20,48 +20,16 @@ package org.apache.skywalking.oap.server.storage.plugin.elasticsearch; import org.apache.skywalking.apm.util.StringUtil; import org.apache.skywalking.oap.server.core.CoreModule; -import org.apache.skywalking.oap.server.core.storage.IBatchDAO; -import org.apache.skywalking.oap.server.core.storage.IHistoryDeleteDAO; -import org.apache.skywalking.oap.server.core.storage.IRegisterLockDAO; -import org.apache.skywalking.oap.server.core.storage.StorageDAO; -import org.apache.skywalking.oap.server.core.storage.StorageException; -import org.apache.skywalking.oap.server.core.storage.StorageModule; -import org.apache.skywalking.oap.server.core.storage.cache.IEndpointInventoryCacheDAO; -import org.apache.skywalking.oap.server.core.storage.cache.INetworkAddressInventoryCacheDAO; -import org.apache.skywalking.oap.server.core.storage.cache.IServiceInstanceInventoryCacheDAO; -import org.apache.skywalking.oap.server.core.storage.cache.IServiceInventoryCacheDAO; -import org.apache.skywalking.oap.server.core.storage.query.IAggregationQueryDAO; -import org.apache.skywalking.oap.server.core.storage.query.IAlarmQueryDAO; -import org.apache.skywalking.oap.server.core.storage.query.IMetadataQueryDAO; -import org.apache.skywalking.oap.server.core.storage.query.IMetricQueryDAO; -import org.apache.skywalking.oap.server.core.storage.query.ITopNRecordsQueryDAO; -import org.apache.skywalking.oap.server.core.storage.query.ITopologyQueryDAO; -import org.apache.skywalking.oap.server.core.storage.query.ITraceQueryDAO; +import org.apache.skywalking.oap.server.core.storage.*; +import org.apache.skywalking.oap.server.core.storage.cache.*; +import org.apache.skywalking.oap.server.core.storage.query.*; import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient; -import org.apache.skywalking.oap.server.library.module.ModuleConfig; -import org.apache.skywalking.oap.server.library.module.ModuleDefine; -import org.apache.skywalking.oap.server.library.module.ModuleProvider; -import org.apache.skywalking.oap.server.library.module.ModuleStartException; -import org.apache.skywalking.oap.server.library.module.ServiceNotProvidedException; -import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.BatchProcessEsDAO; -import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.HistoryDeleteEsDAO; -import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.StorageEsDAO; -import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.StorageEsInstaller; -import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.cache.EndpointInventoryCacheEsDAO; -import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.cache.NetworkAddressInventoryCacheEsDAO; -import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.cache.ServiceInstanceInventoryCacheDAO; -import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.cache.ServiceInventoryCacheEsDAO; -import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.lock.RegisterLockDAOImpl; -import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.lock.RegisterLockInstaller; -import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.query.AggregationQueryEsDAO; -import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.query.AlarmQueryEsDAO; -import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.query.MetadataQueryEsDAO; -import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.query.MetricQueryEsDAO; -import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.query.TopNRecordsQueryEsDAO; -import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.query.TopologyQueryEsDAO; -import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.query.TraceQueryEsDAO; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import org.apache.skywalking.oap.server.library.module.*; +import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.*; +import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.cache.*; +import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.lock.*; +import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.query.*; +import org.slf4j.*; /** * @author peng-yongsheng @@ -117,6 +85,7 @@ public class StorageModuleElasticsearchProvider extends ModuleProvider { this.registerServiceImplementation(IAggregationQueryDAO.class, new AggregationQueryEsDAO(elasticSearchClient)); this.registerServiceImplementation(IAlarmQueryDAO.class, new AlarmQueryEsDAO(elasticSearchClient)); this.registerServiceImplementation(ITopNRecordsQueryDAO.class, new TopNRecordsQueryEsDAO(elasticSearchClient)); + this.registerServiceImplementation(ILogQueryDAO.class, new LogQueryEsDAO(elasticSearchClient)); } @Override diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/LogQueryEsDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/LogQueryEsDAO.java new file mode 100644 index 000000000..c4b0355f1 --- /dev/null +++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/LogQueryEsDAO.java @@ -0,0 +1,105 @@ +/* + * 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.skywalking.oap.server.storage.plugin.elasticsearch.query; + +import com.google.common.base.Strings; +import java.io.IOException; +import java.util.List; +import org.apache.skywalking.oap.server.core.Const; +import org.apache.skywalking.oap.server.core.analysis.manual.log.AbstractLogRecord; +import org.apache.skywalking.oap.server.core.analysis.record.Record; +import org.apache.skywalking.oap.server.core.query.entity.*; +import org.apache.skywalking.oap.server.core.storage.query.ILogQueryDAO; +import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient; +import org.apache.skywalking.oap.server.library.util.BooleanUtils; +import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.EsDAO; +import org.elasticsearch.action.search.SearchResponse; +import org.elasticsearch.index.query.*; +import org.elasticsearch.search.SearchHit; +import org.elasticsearch.search.builder.SearchSourceBuilder; + +import static org.apache.skywalking.oap.server.core.analysis.manual.log.AbstractLogRecord.TRACE_ID; + +/** + * @author wusheng + */ +public class LogQueryEsDAO extends EsDAO implements ILogQueryDAO { + public LogQueryEsDAO(ElasticSearchClient client) { + super(client); + } + + @Override + public Logs queryLogs(String metricName, int serviceId, int serviceInstanceId, int endpointId, + String traceId, LogState state, String stateCode, Pagination paging, int from, int limit, long startSecondTB, + long endSecondTB) throws IOException { + SearchSourceBuilder sourceBuilder = SearchSourceBuilder.searchSource(); + + BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery(); + sourceBuilder.query(boolQueryBuilder); + List mustQueryList = boolQueryBuilder.must(); + + if (startSecondTB != 0 && endSecondTB != 0) { + mustQueryList.add(QueryBuilders.rangeQuery(Record.TIME_BUCKET).gte(startSecondTB).lte(endSecondTB)); + } + + if (serviceId != Const.NONE) { + boolQueryBuilder.must().add(QueryBuilders.termQuery(AbstractLogRecord.SERVICE_ID, serviceId)); + } + if (serviceInstanceId != Const.NONE) { + boolQueryBuilder.must().add(QueryBuilders.termQuery(AbstractLogRecord.SERVICE_INSTANCE_ID, serviceInstanceId)); + } + if (endpointId != Const.NONE) { + boolQueryBuilder.must().add(QueryBuilders.termQuery(AbstractLogRecord.ENDPOINT_ID, endpointId)); + } + if (!Strings.isNullOrEmpty(stateCode)) { + boolQueryBuilder.must().add(QueryBuilders.termQuery(AbstractLogRecord.STATUS_CODE, stateCode)); + } + if (!Strings.isNullOrEmpty(traceId)) { + boolQueryBuilder.must().add(QueryBuilders.termQuery(TRACE_ID, traceId)); + } + if (LogState.ERROR.equals(state)) { + boolQueryBuilder.must().add(QueryBuilders.termQuery(AbstractLogRecord.IS_ERROR, BooleanUtils.booleanToValue(true))); + } else if (LogState.ERROR.equals(state)) { + boolQueryBuilder.must().add(QueryBuilders.termQuery(AbstractLogRecord.IS_ERROR, BooleanUtils.booleanToValue(false))); + } + + sourceBuilder.size(limit); + sourceBuilder.from(from); + + SearchResponse response = getClient().search(metricName, sourceBuilder); + + Logs logs = new Logs(); + logs.setTotal((int)response.getHits().totalHits); + + for (SearchHit searchHit : response.getHits().getHits()) { + Log log = new Log(); + log.setServiceId(((Number)searchHit.getSourceAsMap().get(AbstractLogRecord.SERVICE_ID)).intValue()); + log.setServiceInstanceId(((Number)searchHit.getSourceAsMap().get(AbstractLogRecord.SERVICE_INSTANCE_ID)).intValue()); + log.setEndpointId(((Number)searchHit.getSourceAsMap().get(AbstractLogRecord.ENDPOINT_ID)).intValue()); + log.setError(BooleanUtils.valueToBoolean(((Number)searchHit.getSourceAsMap().get(AbstractLogRecord.IS_ERROR)).intValue())); + log.setStatusCode((String)searchHit.getSourceAsMap().get(AbstractLogRecord.STATUS_CODE)); + log.setContentType(ContentType.instanceOf(((Number)searchHit.getSourceAsMap().get(AbstractLogRecord.CONTENT_TYPE)).intValue())); + log.setContent((String)searchHit.getSourceAsMap().get(AbstractLogRecord.CONTENT)); + + logs.getLogs().add(log); + } + + return logs; + } +} diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/H2StorageProvider.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/H2StorageProvider.java index addd6a0ff..281c30ce0 100644 --- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/H2StorageProvider.java +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/H2StorageProvider.java @@ -87,6 +87,7 @@ public class H2StorageProvider extends ModuleProvider { this.registerServiceImplementation(IAlarmQueryDAO.class, new H2AlarmQueryDAO(h2Client)); this.registerServiceImplementation(IHistoryDeleteDAO.class, new H2HistoryDeleteDAO(h2Client)); this.registerServiceImplementation(ITopNRecordsQueryDAO.class, new H2TopNRecordsQueryDAO(h2Client)); + this.registerServiceImplementation(ILogQueryDAO.class, new H2LogQueryDAO(h2Client)); } @Override public void start() throws ServiceNotProvidedException, ModuleStartException { diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2LogQueryDAO.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2LogQueryDAO.java new file mode 100644 index 000000000..c108c843e --- /dev/null +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2LogQueryDAO.java @@ -0,0 +1,129 @@ +/* + * 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.skywalking.oap.server.storage.plugin.jdbc.h2.dao; + +import com.google.common.base.Strings; +import java.io.IOException; +import java.sql.*; +import java.util.*; +import org.apache.skywalking.oap.server.core.Const; +import org.apache.skywalking.oap.server.core.analysis.manual.log.AbstractLogRecord; +import org.apache.skywalking.oap.server.core.query.entity.*; +import org.apache.skywalking.oap.server.core.storage.query.ILogQueryDAO; +import org.apache.skywalking.oap.server.library.client.jdbc.hikaricp.JDBCHikariCPClient; +import org.apache.skywalking.oap.server.library.util.BooleanUtils; + +import static org.apache.skywalking.oap.server.core.analysis.manual.log.AbstractLogRecord.*; + +/** + * @author wusheng + */ +public class H2LogQueryDAO implements ILogQueryDAO { + private JDBCHikariCPClient h2Client; + + public H2LogQueryDAO(JDBCHikariCPClient h2Client) { + this.h2Client = h2Client; + } + + @Override + public Logs queryLogs(String metricName, int serviceId, int serviceInstanceId, int endpointId, + String traceId, LogState state, + String stateCode, Pagination paging, int from, int limit, long startSecondTB, + long endSecondTB) throws IOException { + StringBuilder sql = new StringBuilder(); + List parameters = new ArrayList<>(10); + + sql.append("from ").append(metricName).append(" where "); + sql.append(" 1=1 "); + if (startSecondTB != 0 && endSecondTB != 0) { + sql.append(" and ").append(AbstractLogRecord.TIME_BUCKET).append(" >= ?"); + parameters.add(startSecondTB); + sql.append(" and ").append(AbstractLogRecord.TIME_BUCKET).append(" <= ?"); + parameters.add(endSecondTB); + } + + if (serviceId != Const.NONE) { + sql.append(" and ").append(SERVICE_ID).append(" = ?"); + parameters.add(serviceId); + } + if (serviceInstanceId != Const.NONE) { + sql.append(" and ").append(AbstractLogRecord.SERVICE_INSTANCE_ID).append(" = ?"); + parameters.add(serviceInstanceId); + } + if (endpointId != Const.NONE) { + sql.append(" and ").append(AbstractLogRecord.ENDPOINT_ID).append(" = ?"); + parameters.add(endpointId); + } + if (!Strings.isNullOrEmpty(stateCode)) { + sql.append(" and ").append(AbstractLogRecord.STATUS_CODE).append(" = ?"); + parameters.add(stateCode); + } + if (!Strings.isNullOrEmpty(traceId)) { + sql.append(" and ").append(TRACE_ID).append(" = ?"); + parameters.add(traceId); + } + if (LogState.ERROR.equals(state)) { + sql.append(" and ").append(AbstractLogRecord.IS_ERROR).append(" = ?"); + parameters.add(BooleanUtils.booleanToValue(true)); + } else if (LogState.ERROR.equals(state)) { + sql.append(" and ").append(AbstractLogRecord.IS_ERROR).append(" = ?"); + parameters.add(BooleanUtils.booleanToValue(false)); + } + + Logs logs = new Logs(); + try (Connection connection = h2Client.getConnection()) { + + try (ResultSet resultSet = h2Client.executeQuery(connection, buildCountStatement(sql.toString()), parameters.toArray(new Object[0]))) { + while (resultSet.next()) { + logs.setTotal(resultSet.getInt("total")); + } + } + + buildLimit(sql, from, limit); + + try (ResultSet resultSet = h2Client.executeQuery(connection, "select * " + sql.toString(), parameters.toArray(new Object[0]))) { + while (resultSet.next()) { + Log log = new Log(); + log.setServiceId(resultSet.getInt(SERVICE_ID)); + log.setServiceInstanceId(resultSet.getInt(SERVICE_INSTANCE_ID)); + log.setEndpointId(resultSet.getInt(ENDPOINT_ID)); + log.setTraceId(resultSet.getString(TRACE_ID)); + log.setTimestamp(resultSet.getString(TIMESTAMP)); + log.setStatusCode(resultSet.getString(STATUS_CODE)); + log.setContentType(ContentType.instanceOf(resultSet.getInt(CONTENT_TYPE))); + log.setContent(resultSet.getString(CONTENT)); + logs.getLogs().add(log); + } + } + } catch (SQLException e) { + throw new IOException(e); + } + + return logs; + } + + protected String buildCountStatement(String sql) { + return "select count(1) total from (select 1 " + sql + " )"; + } + + protected void buildLimit(StringBuilder sql, int from, int limit) { + sql.append(" LIMIT ").append(limit); + sql.append(" OFFSET ").append(from); + } +} diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2TraceQueryDAO.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2TraceQueryDAO.java index ec231e5bc..6b11467cf 100644 --- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2TraceQueryDAO.java +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2TraceQueryDAO.java @@ -103,7 +103,7 @@ public class H2TraceQueryDAO implements ITraceQueryDAO { TraceBrief traceBrief = new TraceBrief(); try (Connection connection = h2Client.getConnection()) { - try (ResultSet resultSet = h2Client.executeQuery(connection, "select count(1) total from (select 1 " + sql.toString() + " )", parameters.toArray(new Object[0]))) { + try (ResultSet resultSet = h2Client.executeQuery(connection, buildCountStatement(sql.toString()), parameters.toArray(new Object[0]))) { while (resultSet.next()) { traceBrief.setTotal(resultSet.getInt("total")); } @@ -132,6 +132,10 @@ public class H2TraceQueryDAO implements ITraceQueryDAO { return traceBrief; } + protected String buildCountStatement(String sql) { + return "select count(1) total from (select 1 " + sql + " )"; + } + protected void buildLimit(StringBuilder sql, int from, int limit) { sql.append(" LIMIT ").append(limit); sql.append(" OFFSET ").append(from); diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLLogQueryDAO.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLLogQueryDAO.java new file mode 100644 index 000000000..d61be023f --- /dev/null +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLLogQueryDAO.java @@ -0,0 +1,40 @@ +/* + * 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.skywalking.oap.server.storage.plugin.jdbc.mysql; + +import org.apache.skywalking.oap.server.library.client.jdbc.hikaricp.JDBCHikariCPClient; +import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.H2LogQueryDAO; + +/** + * @author wusheng + */ +public class MySQLLogQueryDAO extends H2LogQueryDAO { + public MySQLLogQueryDAO(JDBCHikariCPClient h2Client) { + super(h2Client); + } + + protected String buildCountStatement(String sql) { + return "select count(1) total from (select 1 " + sql + " )"; + } + + protected void buildLimit(StringBuilder sql, int from, int limit) { + sql.append(" LIMIT ").append(limit); + sql.append(" OFFSET ").append(from); + } +} diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLStorageProvider.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLStorageProvider.java index 6f2ac9439..9f403e8c1 100644 --- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLStorageProvider.java +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLStorageProvider.java @@ -93,6 +93,7 @@ public class MySQLStorageProvider extends ModuleProvider { this.registerServiceImplementation(IAlarmQueryDAO.class, new MySQLAlarmQueryDAO(mysqlClient)); this.registerServiceImplementation(IHistoryDeleteDAO.class, new H2HistoryDeleteDAO(mysqlClient)); this.registerServiceImplementation(ITopNRecordsQueryDAO.class, new H2TopNRecordsQueryDAO(mysqlClient)); + this.registerServiceImplementation(ILogQueryDAO.class, new MySQLLogQueryDAO(mysqlClient)); } @Override public void start() throws ServiceNotProvidedException, ModuleStartException { diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLTraceQueryDAO.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLTraceQueryDAO.java index 639fe63ca..384d67edc 100644 --- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLTraceQueryDAO.java +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLTraceQueryDAO.java @@ -18,16 +18,8 @@ package org.apache.skywalking.oap.server.storage.plugin.jdbc.mysql; -import com.google.common.base.Strings; -import java.io.IOException; -import java.sql.*; -import java.util.*; -import org.apache.skywalking.oap.server.core.analysis.manual.segment.SegmentRecord; -import org.apache.skywalking.oap.server.core.query.entity.*; import org.apache.skywalking.oap.server.library.client.jdbc.hikaricp.JDBCHikariCPClient; -import org.apache.skywalking.oap.server.library.util.BooleanUtils; import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.H2TraceQueryDAO; -import org.elasticsearch.search.sort.SortOrder; /** * @author wusheng @@ -37,97 +29,8 @@ public class MySQLTraceQueryDAO extends H2TraceQueryDAO { super(mysqlClient); } - @Override - public TraceBrief queryBasicTraces(long startSecondTB, long endSecondTB, long minDuration, long maxDuration, - String endpointName, int serviceId, int serviceInstanceId, int endpointId, String traceId, int limit, int from, - TraceState traceState, QueryOrder queryOrder) throws IOException { - StringBuilder sql = new StringBuilder(); - List parameters = new ArrayList<>(10); - - sql.append("from ").append(SegmentRecord.INDEX_NAME).append(" where "); - sql.append(" 1=1 "); - if (startSecondTB != 0 && endSecondTB != 0) { - sql.append(" and ").append(SegmentRecord.TIME_BUCKET).append(" >= ?"); - parameters.add(startSecondTB); - sql.append(" and ").append(SegmentRecord.TIME_BUCKET).append(" <= ?"); - parameters.add(endSecondTB); - } - if (minDuration != 0 || maxDuration != 0) { - if (minDuration != 0) { - sql.append(" and ").append(SegmentRecord.LATENCY).append(" >= ?"); - parameters.add(minDuration); - } - if (maxDuration != 0) { - sql.append(" and ").append(SegmentRecord.LATENCY).append(" <= ?"); - parameters.add(maxDuration); - } - } - if (!Strings.isNullOrEmpty(endpointName)) { - sql.append(" and ").append(SegmentRecord.ENDPOINT_NAME).append(" like '%" + endpointName + "%'"); - } - if (serviceId != 0) { - sql.append(" and ").append(SegmentRecord.SERVICE_ID).append(" = ?"); - parameters.add(serviceId); - } - if (serviceInstanceId != 0) { - sql.append(" and ").append(SegmentRecord.SERVICE_INSTANCE_ID).append(" = ?"); - parameters.add(serviceInstanceId); - } - if (endpointId != 0) { - sql.append(" and ").append(SegmentRecord.ENDPOINT_ID).append(" = ?"); - parameters.add(endpointId); - } - if (!Strings.isNullOrEmpty(traceId)) { - sql.append(" and ").append(SegmentRecord.TRACE_ID).append(" = ?"); - parameters.add(traceId); - } - switch (traceState) { - case ERROR: - sql.append(" and ").append(SegmentRecord.IS_ERROR).append(" = ").append(BooleanUtils.TRUE); - break; - case SUCCESS: - sql.append(" and ").append(SegmentRecord.IS_ERROR).append(" = ").append(BooleanUtils.FALSE); - break; - } - switch (queryOrder) { - case BY_START_TIME: - sql.append(" order by ").append(SegmentRecord.START_TIME).append(" ").append(SortOrder.DESC); - break; - case BY_DURATION: - sql.append(" order by ").append(SegmentRecord.LATENCY).append(" ").append(SortOrder.DESC); - break; - } - - TraceBrief traceBrief = new TraceBrief(); - try (Connection connection = getClient().getConnection()) { - - try (ResultSet resultSet = getClient().executeQuery(connection, "select count(1) total from (select 1 " + sql.toString() + " ) AS TRACE", parameters.toArray(new Object[0]))) { - while (resultSet.next()) { - traceBrief.setTotal(resultSet.getInt("total")); - } - } - - buildLimit(sql, from, limit); - - try (ResultSet resultSet = getClient().executeQuery(connection, "select * " + sql.toString(), parameters.toArray(new Object[0]))) { - while (resultSet.next()) { - BasicTrace basicTrace = new BasicTrace(); - - basicTrace.setSegmentId(resultSet.getString(SegmentRecord.SEGMENT_ID)); - basicTrace.setStart(resultSet.getString(SegmentRecord.START_TIME)); - basicTrace.getEndpointNames().add(resultSet.getString(SegmentRecord.ENDPOINT_NAME)); - basicTrace.setDuration(resultSet.getInt(SegmentRecord.LATENCY)); - basicTrace.setError(BooleanUtils.valueToBoolean(resultSet.getInt(SegmentRecord.IS_ERROR))); - String traceIds = resultSet.getString(SegmentRecord.TRACE_ID); - basicTrace.getTraceIds().add(traceIds); - traceBrief.getTraces().add(basicTrace); - } - } - } catch (SQLException e) { - throw new IOException(e); - } - - return traceBrief; + @Override protected String buildCountStatement(String sql) { + return "select count(1) total from (select 1 " + sql + " ) AS TRACE"; } @Override protected void buildLimit(StringBuilder sql, int from, int limit) {