diff --git a/CHANGES.md b/CHANGES.md index ec780510f..d90a2f05b 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -12,6 +12,7 @@ Release Notes. #### OAP-Backend * Make meter receiver support MAL. +* Support influxDB connection response format option. Fix some error when use JSON as influxDB response format. * Support Kafka MirrorMaker 2.0 to replicate topics between Kafka clusters. #### UI diff --git a/docs/en/setup/backend/configuration-vocabulary.md b/docs/en/setup/backend/configuration-vocabulary.md index 1b10d1a5f..d6c5717db 100644 --- a/docs/en/setup/backend/configuration-vocabulary.md +++ b/docs/en/setup/backend/configuration-vocabulary.md @@ -140,6 +140,7 @@ core|default|role|Option values, `Mixed/Receiver/Aggregator`. **Receiver** mode | - | - | duration | The time to wait at most (milliseconds). | SW_STORAGE_INFLUXDB_DURATION | 1000| | - | - | batchEnabled | If true, write points with batch api. | SW_STORAGE_INFLUXDB_BATCH_ENABLED | true| | - | - | fetchTaskLogMaxSize | The max number of fetch task log in a request. | SW_STORAGE_INFLUXDB_FETCH_TASK_LOG_MAX_SIZE | 5000| +| - | - | connectionResponseFormat | The response format of connection to influxDB, cannot be anything but MSGPACK or JSON. | SW_STORAGE_INFLUXDB_CONNECTION_RESPONSE_FORMAT | MSGPACK | | agent-analyzer | default | Agent Analyzer. | SW_AGENT_ANALYZER | default | | - | -| sampleRate|Sampling rate for receiving trace. The precision is 1/10000. 10000 means 100% sample in default.|SW_TRACE_SAMPLE_RATE|10000| | - | - |slowDBAccessThreshold|The slow database access thresholds. Unit ms.|SW_SLOW_DB_THRESHOLD|default:200,mongodb:100| diff --git a/oap-server/server-bootstrap/src/main/resources/application.yml b/oap-server/server-bootstrap/src/main/resources/application.yml index d30daca70..ca2723571 100755 --- a/oap-server/server-bootstrap/src/main/resources/application.yml +++ b/oap-server/server-bootstrap/src/main/resources/application.yml @@ -194,6 +194,7 @@ storage: duration: ${SW_STORAGE_INFLUXDB_DURATION:1000} # the time to wait at most (milliseconds) batchEnabled: ${SW_STORAGE_INFLUXDB_BATCH_ENABLED:true} fetchTaskLogMaxSize: ${SW_STORAGE_INFLUXDB_FETCH_TASK_LOG_MAX_SIZE:5000} # the max number of fetch task log in a request + connectionResponseFormat: ${SW_STORAGE_INFLUXDB_CONNECTION_RESPONSE_FORMAT:MSGPACK} # the response format of connection to influxDB, cannot be anything but MSGPACK or JSON. agent-analyzer: selector: ${SW_AGENT_ANALYZER:default} diff --git a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/InfluxClient.java b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/InfluxClient.java index f49ee3ee4..a2be5695f 100644 --- a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/InfluxClient.java +++ b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/InfluxClient.java @@ -72,10 +72,11 @@ public class InfluxClient implements Client, HealthCheckable { @Override public void connect() { try { + InfluxDB.ResponseFormat responseFormat = InfluxDB.ResponseFormat.valueOf(config.getConnectionResponseFormat()); influx = InfluxDBFactory.connect(config.getUrl(), config.getUser(), config.getPassword(), new OkHttpClient.Builder().readTimeout(3, TimeUnit.MINUTES) .writeTimeout(3, TimeUnit.MINUTES), - InfluxDB.ResponseFormat.MSGPACK + responseFormat ); influx.query(new Query("CREATE DATABASE " + database)); influx.enableGzip(); diff --git a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/InfluxStorageConfig.java b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/InfluxStorageConfig.java index 19f99e3fc..d287f8c38 100644 --- a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/InfluxStorageConfig.java +++ b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/InfluxStorageConfig.java @@ -35,4 +35,5 @@ public class InfluxStorageConfig extends ModuleConfig { private boolean batchEnabled = true; private int fetchTaskLogMaxSize = 5000; + private String connectionResponseFormat = "MSGPACK"; } diff --git a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/ProfileThreadSnapshotQuery.java b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/ProfileThreadSnapshotQuery.java index a1a7b5224..b5d08be77 100644 --- a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/ProfileThreadSnapshotQuery.java +++ b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/ProfileThreadSnapshotQuery.java @@ -96,7 +96,7 @@ public class ProfileThreadSnapshotQuery implements IProfileThreadSnapshotQueryDA BasicTrace basicTrace = new BasicTrace(); basicTrace.setSegmentId((String) values.get(2)); - basicTrace.setStart(String.valueOf(values.get(3))); + basicTrace.setStart(String.valueOf(((Number) values.get(3)).longValue())); basicTrace.getEndpointNames().add((String) values.get(4)); basicTrace.setDuration(((Number) values.get(5)).intValue()); basicTrace.setError(BooleanUtils.valueToBoolean(((Number) values.get(6)).intValue())); @@ -190,11 +190,11 @@ public class ProfileThreadSnapshotQuery implements IProfileThreadSnapshotQueryDA segmentRecord.setTraceId((String) values.get(2)); segmentRecord.setServiceId((String) values.get(3)); segmentRecord.setEndpointName((String) values.get(4)); - segmentRecord.setStartTime((long) values.get(5)); - segmentRecord.setEndTime((long) values.get(6)); - segmentRecord.setLatency((int) values.get(7)); - segmentRecord.setIsError((int) values.get(8)); - segmentRecord.setVersion((int) values.get(10)); + segmentRecord.setStartTime(((Number) values.get(5)).longValue()); + segmentRecord.setEndTime(((Number) values.get(6)).longValue()); + segmentRecord.setLatency(((Number) values.get(7)).intValue()); + segmentRecord.setIsError(((Number) values.get(8)).intValue()); + segmentRecord.setVersion(((Number) values.get(10)).intValue()); String base64 = (String) values.get(9); if (!Strings.isNullOrEmpty(base64)) { diff --git a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/TraceQuery.java b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/TraceQuery.java index f01602b92..2b969c2c1 100644 --- a/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/TraceQuery.java +++ b/oap-server/server-storage-plugin/storage-influxdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/influxdb/query/TraceQuery.java @@ -165,7 +165,7 @@ public class TraceQuery implements ITraceQueryDAO { BasicTrace basicTrace = new BasicTrace(); basicTrace.setSegmentId((String) values.get(2)); - basicTrace.setStart(String.valueOf(values.get(3))); + basicTrace.setStart(String.valueOf(((Number) values.get(3)).longValue())); basicTrace.getEndpointNames().add((String) values.get(4)); basicTrace.setDuration(((Number) values.get(5)).intValue()); basicTrace.setError(BooleanUtils.valueToBoolean(((Number) values.get(6)).intValue()));