集成skywalking

iot-common/iot-base/src/main/java/com/qiuguo/iot/base/constans/Log4Constans.java

@@ -1,6 +1,9 @@
 package com.qiuguo.iot.base.constans;
 
 public class Log4Constans {
-    public static String PRINT_LOG_ID = "tid";
+    public static String PRINT_LOG_ID = "traceId";
+    public static String HEADER_TRACE_ID = "x-trace-id";
+
+
     public static String CUSTOMER_IP = "customerIp";
 }

iot-gateway/pom.xml

@@ -71,6 +71,18 @@
             <version>0.0.1-SNAPSHOT</version>
             <scope>compile</scope>
         </dependency>
+        <!-- https://mvnrepository.com/artifact/org.apache.skywalking/apm-toolkit-trace -->
+        <dependency>
+            <groupId>org.apache.skywalking</groupId>
+            <artifactId>apm-toolkit-trace</artifactId>
+            <version>8.15.0</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.skywalking</groupId>
+            <artifactId>apm-toolkit-webflux</artifactId>
+            <version>8.15.0</version>
+        </dependency>
     
     </dependencies>
 

iot-gateway/src/main/java/com/qiuguo/iot/gateway/filter/AuthFilter.java

@@ -8,7 +8,12 @@ import com.qiuguo.iot.gateway.config.properties.XssProperties;
 
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
+import java.util.Optional;
+
 import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.apm.toolkit.trace.Trace;
+import org.apache.skywalking.apm.toolkit.trace.TraceContext;
+import org.apache.skywalking.apm.toolkit.webflux.WebFluxSkyWalkingOperators;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.cloud.gateway.filter.GatewayFilterChain;
 import org.springframework.cloud.gateway.filter.GlobalFilter;
@@ -43,16 +48,19 @@ public class AuthFilter implements GlobalFilter, Ordered {
 
     
     @Override
+    @Trace
     public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
 
+        String traceId = WebFluxSkyWalkingOperators.continueTracing(exchange, TraceContext::traceId);
+        //Optional<String> ss = TraceContext.getCorrelation("traceId");
         ServerHttpRequest request = exchange.getRequest();
         String url = request.getPath().toString();
         String customerIp = request.getRemoteAddress().getAddress().getHostAddress();
-        String tracId = StringUtils.getUUID();
         ServerWebExchange.Builder ex = exchange.mutate();
-        ex.request(getRequest(exchange, customerIp, tracId));
+        ex.request(getRequest(exchange, customerIp, traceId));
+        ServerWebExchange newEx = ex.build();
         if (xssProperties.getExcludeUrls().contains(url)) {
-            return chain.filter(ex.build());
+            return chain.filter(newEx);
         }
 
         String api_token = exchange.getRequest().getHeaders().getFirst(UserAuthContains.API_TOKEN);
@@ -60,7 +68,9 @@ public class AuthFilter implements GlobalFilter, Ordered {
         if (ObjectUtils.isEmpty(api_token) || ObjectUtils.isEmpty(api_type)) {
             return Mono.error(new RuntimeException("未登录"));
         }
-        return chain.filter(ex.build());
+        return chain.filter(newEx);
+
+
         // String key = RedisConstans.IOT_TOKEN.concat(api_token);
         // return reactiveRedisTemplate.getExpire(key).map(Duration::getSeconds).flatMap(ttl -> {
         //     if (ttl == -1) {
@@ -78,13 +88,15 @@ public class AuthFilter implements GlobalFilter, Ordered {
 
     private ServerHttpRequest getRequest(ServerWebExchange exchange, String customerIp, String tracId){
         ServerHttpRequest request = exchange.getRequest();
+
+       exchange.getResponse().getHeaders().set(Log4Constans.HEADER_TRACE_ID, tracId);
         ServerHttpRequest newRequest = new ServerHttpRequestDecorator(request){
             @Override
             public HttpHeaders getHeaders(){
                 HttpHeaders httpHeaders = new HttpHeaders();
                 httpHeaders.putAll(super.getHeaders());
                 httpHeaders.set(Log4Constans.CUSTOMER_IP, customerIp);
-                httpHeaders.set(Log4Constans.PRINT_LOG_ID, tracId);
+                httpHeaders.set(Log4Constans.HEADER_TRACE_ID, tracId);
                 return httpHeaders;
             }
         };

iot-modules/iot-box-user-api/pom.xml

@@ -70,11 +70,16 @@
             <artifactId>reactor-test</artifactId>
             <scope>test</scope>
         </dependency>
-
         <dependency>
             <groupId>org.apache.skywalking</groupId>
             <artifactId>apm-toolkit-logback-1.x</artifactId>
-            <version>8.7.0</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>ch.qos.logback</groupId>
+                    <artifactId>logback-classic</artifactId>
+                </exclusion>
+            </exclusions>
+            <version>8.15.0</version>
         </dependency>
         <!--skywalking trace end-->
 
@@ -83,6 +88,19 @@
             <artifactId>hutool-all</artifactId>
             <version>5.8.21</version>
         </dependency>
+
+        <!-- https://mvnrepository.com/artifact/org.apache.skywalking/apm-toolkit-trace -->
+        <dependency>
+            <groupId>org.apache.skywalking</groupId>
+            <artifactId>apm-toolkit-trace</artifactId>
+            <version>8.15.0</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.skywalking</groupId>
+            <artifactId>apm-toolkit-webflux</artifactId>
+            <version>8.15.0</version>
+        </dependency>
     </dependencies>
     
     <build>

iot-modules/iot-box-user-api/src/main/java/com/qiuguo/iot/user/api/controller/device/DeviceController.java

@@ -23,6 +23,10 @@ import com.qiuguo.iot.data.service.device.DeviceUserTalkRecordService;
 import com.qiuguo.iot.third.service.TuyaDeviceConnector;
 import com.qiuguo.iot.data.resp.device.DeviceInitResp;
 import com.qiuguo.iot.user.api.rest.device.SetDeviceBindInfoRest;
+import org.apache.skywalking.apm.toolkit.trace.CarrierItemRef;
+import org.apache.skywalking.apm.toolkit.trace.ContextCarrierRef;
+import org.apache.skywalking.apm.toolkit.trace.Tracer;
+import org.apache.skywalking.apm.toolkit.webflux.WebFluxSkyWalkingOperators;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.data.redis.core.ReactiveStringRedisTemplate;
@@ -72,6 +76,7 @@ public class DeviceController {
     public Mono<DeviceInitResp> deviceInit(@RequestParam String wifiMac, @RequestParam String btMac,
                                            @RequestParam Integer type, @RequestParam Long time,
                                            @RequestParam String signature){
+
         //deviceInfoService.
         Long now = System.currentTimeMillis();
         if(checkTimeout && now - time > timeOut){
@@ -95,7 +100,8 @@ public class DeviceController {
         if (!md5.equals(signature)) {
             //
             //验签失败
-            log.info("验签失败");
+
+            log.info("验签失败，{}", md5);
             BusinessException ex = new BusinessException("验签失败");
             return Mono.error(ex);
         }
@@ -103,9 +109,8 @@ public class DeviceController {
         DeviceInfoRequest request = new DeviceInfoRequest();
         request.setWifiMac(wifiMac);
         request.setBtMac(btMac);
-        Mono<DeviceInfoEntity> mono = deviceInfoService.selectDeviceInfoByRequest(request);
 
-        return mono.defaultIfEmpty(new DeviceInfoEntity()).flatMap(entity -> {
+        return deviceInfoService.selectDeviceInfoByRequest(request).defaultIfEmpty(new DeviceInfoEntity()).flatMap(entity -> {
             Mono mono1 = null;
             if(entity.getId() == null){
                 entity.setWifiMac(wifiMac);

iot-modules/iot-box-user-api/src/main/java/com/qiuguo/iot/user/api/filter/LogMdcConfiguration.java

@@ -2,6 +2,10 @@ package com.qiuguo.iot.user.api.filter;
 
 import com.qiuguo.iot.base.constans.Log4Constans;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.apm.toolkit.trace.Trace;
+import org.apache.skywalking.apm.toolkit.trace.TraceContext;
+import org.apache.skywalking.apm.toolkit.trace.Tracer;
+import org.apache.skywalking.apm.toolkit.webflux.WebFluxSkyWalkingOperators;
 import org.reactivestreams.Subscription;
 import org.slf4j.MDC;
 import org.springframework.context.annotation.Configuration;
@@ -22,14 +26,18 @@ import javax.annotation.PreDestroy;
 @Configuration
 @Slf4j
 public class LogMdcConfiguration {
+    public int hashcode;
     @PostConstruct
     public void contextOperatorHook() {
         Hooks.onEachOperator(Log4Constans.PRINT_LOG_ID, Operators.lift((r, c) ->{
             Context ctx = c.currentContext();
+
             if(ctx.hasKey(Log4Constans.PRINT_LOG_ID)){
                 MDC.put(Log4Constans.PRINT_LOG_ID, ctx.get(Log4Constans.PRINT_LOG_ID));
+                Tracer.createLocalSpan(Log4Constans.PRINT_LOG_ID);
             }
             return new MdcContextSubscriber(c);
+
         }));
     }
     @PreDestroy
@@ -46,19 +54,28 @@ public class LogMdcConfiguration {
         @Override
         public void onComplete() {
             coreSubscriber.onComplete();
+            Tracer.stopSpan();
             MDC.remove(Log4Constans.PRINT_LOG_ID);
+
         }
 
         @Override
         public void onError(Throwable throwable) {
-            log.info("异常{}", throwable);
+            int hcode = throwable.hashCode();
-            coreSubscriber.onError(throwable);
+            if(hcode != hashcode){
+                log.info("异常{}", throwable);
+                hashcode = hcode;
+            }
+
             MDC.remove(Log4Constans.PRINT_LOG_ID);
+            Tracer.stopSpan();
+            coreSubscriber.onError(throwable);
 
         }
 
         @Override
         public void onSubscribe(Subscription subscription) {
+
             coreSubscriber.onSubscribe(subscription);
         }
 

iot-modules/iot-box-user-api/src/main/java/com/qiuguo/iot/user/api/filter/LogWebFilter.java

@@ -2,14 +2,18 @@ package com.qiuguo.iot.user.api.filter;
 
 import com.qiuguo.iot.base.constans.Log4Constans;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.apm.toolkit.trace.*;
+import org.apache.skywalking.apm.toolkit.webflux.WebFluxSkyWalkingOperators;
 import org.hswebframework.web.logger.ReactiveLogger;
 import org.reactivestreams.Publisher;
 import org.slf4j.MDC;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.core.annotation.Order;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferFactory;
 import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.core.io.buffer.DefaultDataBufferFactory;
+import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpMethod;
 import org.springframework.http.server.reactive.ServerHttpRequest;
 import org.springframework.http.server.reactive.ServerHttpRequestDecorator;
@@ -18,8 +22,10 @@ import org.springframework.http.server.reactive.ServerHttpResponseDecorator;
 import org.springframework.web.server.ServerWebExchange;
 import org.springframework.web.server.WebFilter;
 import org.springframework.web.server.WebFilterChain;
+import org.springframework.web.servlet.ModelAndView;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.publisher.SignalType;
 import reactor.util.context.Context;
 
 import java.nio.charset.StandardCharsets;
@@ -35,15 +41,15 @@ import java.nio.charset.StandardCharsets;
 public class LogWebFilter implements WebFilter {
     @Override
     public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
+        //Tracer.stopSpan();
         long startTime = System.currentTimeMillis();
         ServerHttpRequest request = exchange.getRequest();
         String requestId = request.getId();
         String ip = request.getRemoteAddress().getAddress().getHostAddress();
-        if(request.getHeaders().containsKey(Log4Constans.PRINT_LOG_ID)){
+        if(request.getHeaders().containsKey(Log4Constans.HEADER_TRACE_ID)){
             //网关生成的tracId
-            requestId = request.getHeaders().get(Log4Constans.PRINT_LOG_ID).get(0);
+            requestId = request.getHeaders().get(Log4Constans.HEADER_TRACE_ID).get(0);
         }
-
         if(request.getHeaders().containsKey(Log4Constans.CUSTOMER_IP)){
             //网关透传过来的IP
             ip = request.getHeaders().get(Log4Constans.CUSTOMER_IP).get(0);
@@ -53,7 +59,6 @@ public class LogWebFilter implements WebFilter {
 
 
         String m = request.getMethod().toString();
-
         log.info("api start time:{} ip:{} method:{} url:{} param:{} headers:{}",
                 startTime,
                 ip,
@@ -66,36 +71,45 @@ public class LogWebFilter implements WebFilter {
 
         ex.response(getResponse(exchange, requestId));
 
-        if(!request.getMethod().equals(HttpMethod.GET) && !request.getMethod().equals(HttpMethod.DELETE)){
+
-            ex.request(getRequest(exchange));
+        ex.request(getRequest(exchange));
-        }
         String tracId = requestId;
-        return chain.filter(ex.build())
+        ServerWebExchange newEx = ex.build();
+
+        return chain.filter(newEx)
                 .contextWrite(context -> {
                     Context contextTmp = context.put(Log4Constans.PRINT_LOG_ID, tracId);
-
                     return contextTmp;
-                })
+
-                .doFinally(signalType -> {
+                }).doFinally(signalType -> {
                     long endTime = System.currentTimeMillis();
                     MDC.put(Log4Constans.PRINT_LOG_ID, tracId);
                     log.info("api end time:{}, total time:{}", endTime, endTime - startTime);
                     MDC.remove(Log4Constans.PRINT_LOG_ID);
                 });
+
+
+
+
+
     }
 
     private ServerHttpRequest getRequest(ServerWebExchange exchange){
+
         ServerHttpRequest request = exchange.getRequest();
         return new ServerHttpRequestDecorator(request){
             @Override
             public Flux<DataBuffer> getBody() {
                 Flux<DataBuffer> body = this.getDelegate().getBody();
                 return body.map(dataBuffer -> {
-                    log.info("request:{}", dataBuffer.toString(StandardCharsets.UTF_8));
+                    if(!request.getMethod().equals(HttpMethod.GET) && !request.getMethod().equals(HttpMethod.DELETE)) {
+                        log.info("request:{}", dataBuffer.toString(StandardCharsets.UTF_8));
+                    }
                     return dataBuffer;
                 });
             }
         };
+
     }
 
     private ServerHttpResponse getResponse(ServerWebExchange exchange, String requestId){
@@ -126,6 +140,7 @@ public class LogWebFilter implements WebFilter {
                 return super.writeWith(body);
             }
         };
+
     }
 
 }

iot-modules/iot-box-user-api/src/main/java/com/qiuguo/iot/user/api/filter/LogbackMDCPatternConverter.java

@@ -1,15 +0,0 @@
-package com.qiuguo.iot.user.api.filter;
-
-import ch.qos.logback.classic.pattern.MDCConverter;
-import ch.qos.logback.classic.spi.ILoggingEvent;
-
-public class LogbackMDCPatternConverter extends MDCConverter {
-    @Override
-    public void start() {
-        super.start();
-    }
-    @Override
-    public String convert(ILoggingEvent iLoggingEvent) {
-        return super.convert(iLoggingEvent);
-    }
-}

iot-modules/iot-box-user-api/src/main/java/com/qiuguo/iot/user/api/filter/TraceIdMDCPatternLogbackLayout.java

@@ -1,10 +0,0 @@
-package com.qiuguo.iot.user.api.filter;
-
-import ch.qos.logback.classic.PatternLayout;
-
-public class TraceIdMDCPatternLogbackLayout extends PatternLayout {
-    static {
-        defaultConverterMap.put("X", LogbackMDCPatternConverter.class.getName());
-        defaultConverterMap.put("mdc", LogbackMDCPatternConverter.class.getName());
-    }
-}

iot-modules/iot-box-user-api/src/main/resources/logback-spring.xml

@@ -38,7 +38,7 @@
 
     <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
         <encoder class="ch.qos.logback.core.encoder.LayoutWrappingEncoder">
-            <layout class="com.qiuguo.iot.user.api.filter.TraceIdMDCPatternLogbackLayout">
+            <layout class="org.apache.skywalking.apm.toolkit.log.logback.v1.x.mdc.TraceIdMDCPatternLogbackLayout">
                 <Pattern>${PATTERN}</Pattern>
             </layout>
         </encoder>
@@ -47,7 +47,7 @@
     <!--  skywalking采集日志  -->
     <appender name="SKYWALKING" class="org.apache.skywalking.apm.toolkit.log.logback.v1.x.log.GRPCLogClientAppender">
         <encoder class="ch.qos.logback.core.encoder.LayoutWrappingEncoder">
-            <layout class="com.qiuguo.iot.user.api.filter.TraceIdMDCPatternLogbackLayout">
+            <layout class="org.apache.skywalking.apm.toolkit.log.logback.v1.x.mdc.TraceIdMDCPatternLogbackLayout">
                 <Pattern>${SKY_PATTERN}</Pattern>
             </layout>
         </encoder>