Merge branch 'hai/ip-restriction' into bard/add_token_info_into_tx_20…

…240424

Makefile

@@ -16,16 +16,12 @@ DOCKER_COMPOSE_BRIDGE := xlayer-bridge-service
 DOCKER_COMPOSE_BRIDGE_V1TOV2 := xlayer-bridge-service-v1tov2
 
 DOCKER_COMPOSE_REDIS := xlayer-bridge-redis
-DOCKER_COMPOSE_COIN_KAFKA_NODE := xlayer-bridge-coin-kafka
-DOCKER_COMPOSE_ZOOKEEPER := kafka-zookeeper
 
 RUN_STATE_DB := $(DOCKER_COMPOSE) up -d $(DOCKER_COMPOSE_STATE_DB)
 RUN_POOL_DB := $(DOCKER_COMPOSE) up -d $(DOCKER_COMPOSE_POOL_DB)
 RUN_BRIDGE_DB := $(DOCKER_COMPOSE) up -d $(DOCKER_COMPOSE_BRIDGE_DB)
 RUN_REDIS := $(DOCKER_COMPOSE) up -d $(DOCKER_COMPOSE_REDIS)
-RUN_COIN_KAFKA := $(DOCKER_COMPOSE) up -d $(DOCKER_COMPOSE_COIN_KAFKA_NODE)
-RUN_ZOOKEEPER := $(DOCKER_COMPOSE) up -d $(DOCKER_COMPOSE_ZOOKEEPER)
-RUN_DBS := ${RUN_BRIDGE_DB} && ${RUN_STATE_DB} && ${RUN_POOL_DB} && ${RUN_REDIS} && ${RUN_COIN_KAFKA} && ${RUN_ZOOKEEPER}
+RUN_DBS := ${RUN_BRIDGE_DB} && ${RUN_STATE_DB} && ${RUN_POOL_DB} && ${RUN_REDIS}
 RUN_NODE := $(DOCKER_COMPOSE) up -d $(DOCKER_COMPOSE_ZKEVM_NODE)
 RUN_NODE_V1TOV2 := $(DOCKER_COMPOSE) up -d $(DOCKER_COMPOSE_ZKEVM_NODE_V1TOV2)
 RUN_AGGREGATOR_V1TOV2 := $(DOCKER_COMPOSE) up -d $(DOCKER_COMPOSE_ZKEVM_AGGREGATOR_V1TOV2)

cmd/run_xlayer.go

@@ -302,21 +302,23 @@ func startServer(ctx *cli.Context, opts ...runOptionFunc) error {
 		// init token logo client
 		tokenlogoinfo.InitClient(c.TokenLogoServiceConfig)
 
-		// Start the coin middleware kafka consumer
-		log.Debugf("start initializing kafka consumer...")
-		coinKafkaConsumer, err := coinmiddleware.NewKafkaConsumer(c.CoinKafkaConsumer, redisStorage)
-		if err != nil {
-			log.Error(err)
-			return err
-		}
-		log.Debugf("finish initializing kafka consumer")
-		go coinKafkaConsumer.Start(ctx.Context)
-		defer func() {
-			err := coinKafkaConsumer.Close()
+		if len(c.CoinKafkaConsumer.Brokers) > 0 {
+			// Start the coin middleware kafka consumer
+			log.Debugf("start initializing kafka consumer...")
+			coinKafkaConsumer, err := coinmiddleware.NewKafkaConsumer(c.CoinKafkaConsumer, redisStorage)
 			if err != nil {
-				log.Errorf("close kafka consumer error: %v", err)
+				log.Error(err)
+				return err
 			}
-		}()
+			log.Debugf("finish initializing kafka consumer")
+			go coinKafkaConsumer.Start(ctx.Context)
+			defer func() {
+				err := coinKafkaConsumer.Close()
+				if err != nil {
+					log.Errorf("close kafka consumer error: %v", err)
+				}
+			}()
+		}
 	}
 
 	// Wait for an in interrupt.

config/config.debug.toml

@@ -68,7 +68,7 @@ BridgeVersion = "v1"
     MockPrice = true
 
 [CoinKafkaConsumer]
-Brokers = ["localhost:9092"]
+Brokers = []
 Topics = ["explorer_chainAddressPrice_push"]
 ConsumerGroupID = "xlayer-bridge-service"
 InitialOffset = -1

config/config.local.toml

@@ -69,7 +69,7 @@ SentinelConfigFilePath = "/app/sentinel_config.json"
     MockPrice = true
 
 [CoinKafkaConsumer]
-Brokers = ["xlayer-bridge-coin-kafka:9092"]
+Brokers = []
 Topics = ["explorer_chainAddressPrice_push"]
 ConsumerGroupID = "xlayer-bridge-service"
 InitialOffset = -1

docker-compose.yml

@@ -77,26 +77,26 @@ services:
       - REDIS_PORT=6379
       - REDIS_DATABASES=8
 
-  kafka-zookeeper:
-    image: wurstmeister/zookeeper
-    container_name: kafka-zookeeper
-    ports:
-      - "2181:2181"
-    expose:
-      - 2181
+#  kafka-zookeeper:
+#    image: wurstmeister/zookeeper
+#    container_name: kafka-zookeeper
+#    ports:
+#      - "2181:2181"
+#    expose:
+#      - 2181
 
-  xlayer-bridge-coin-kafka:
-    image: wurstmeister/kafka
-    container_name: xlayer-bridge-coin-kafka
-    expose:
-      - 9092
-    environment:
-      KAFKA_ADVERTISED_LISTENERS: INSIDE://:9092,OUTSIDE://localhost:9123
-      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INSIDE:PLAINTEXT,OUTSIDE:PLAINTEXT
-      KAFKA_LISTENERS: INSIDE://0.0.0.0:9092,OUTSIDE://0.0.0.0:9123
-      KAFKA_INTER_BROKER_LISTENER_NAME: INSIDE
-      KAFKA_ZOOKEEPER_CONNECT: kafka-zookeeper:2181
-      KAFKA_CREATE_TOPICS: "explorer_chainAddressPrice_push:1:1"
+#  xlayer-bridge-coin-kafka:
+#    image: wurstmeister/kafka
+#    container_name: xlayer-bridge-coin-kafka
+#    expose:
+#      - 9092
+#    environment:
+#      KAFKA_ADVERTISED_LISTENERS: INSIDE://:9092,OUTSIDE://localhost:9123
+#      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INSIDE:PLAINTEXT,OUTSIDE:PLAINTEXT
+#      KAFKA_LISTENERS: INSIDE://0.0.0.0:9092,OUTSIDE://0.0.0.0:9123
+#      KAFKA_INTER_BROKER_LISTENER_NAME: INSIDE
+#      KAFKA_ZOOKEEPER_CONNECT: kafka-zookeeper:2181
+#      KAFKA_CREATE_TOPICS: "explorer_chainAddressPrice_push:1:1"
 
   xlayer-bridge-service-api:
     container_name: xlayer-bridge-service-api

proto/src/proto/bridge/v1/query.proto

@@ -262,6 +262,7 @@ enum ErrorCode {
     ERROR_OK = 0;
     // Start from 1000 so that it will not conflict with gRPC error code range
     ERROR_DEFAULT = 1000;
+    ERROR_BUSINESS = 1001; // General business error
     ERROR_IP_RESTRICTED = 1009;
 }
 

server/interceptors.go

@@ -68,12 +68,12 @@ func NewRequestMetricsInterceptor() grpc.UnaryServerInterceptor {
 // getRespErrorInfo returns the error code and msg from the resp
 func getRespErrorInfo(resp any, err error) (code int64, msg string) {
 	if err != nil {
-		return defaultErrorCode, err.Error()
+		return int64(pb.ErrorCode_ERROR_DEFAULT), err.Error()
 	}
 
 	if resp == nil {
 		// This should not happen
-		return defaultSuccessCode, ""
+		return int64(pb.ErrorCode_ERROR_OK), ""
 	}
 
 	// Check `Msg" field in the resp body

server/interceptors_test.go

@@ -20,14 +20,14 @@ func TestGetRespErrorInfo(t *testing.T) {
 			"err is not nil",
 			pb.CommonCoinsResponse{Code: 2, Msg: "this will not be returned"},
 			errors.New("this is an error"),
-			defaultErrorCode,
+			int64(pb.ErrorCode_ERROR_DEFAULT),
 			"this is an error",
 		},
 		{
 			"no error and resp is nil",
 			nil,
 			nil,
-			defaultSuccessCode,
+			int64(pb.ErrorCode_ERROR_OK),
 			"",
 		},
 		{
@@ -41,7 +41,7 @@ func TestGetRespErrorInfo(t *testing.T) {
 			"no code and msg field",
 			struct{}{},
 			nil,
-			defaultSuccessCode,
+			int64(pb.ErrorCode_ERROR_OK),
 			"",
 		},
 	}