add new kafka for post receipt

cmd/producer/producer.go

@@ -67,7 +67,7 @@ func produceTask() {
 		select {
 		case <-ticker.C:
 			task := makeTask()
-			if err := Fire(producer, task, config.GetConfig().Kafka.Topic); err != nil {
+			if err := Fire(producer, task, config.GetConfig().Kafka.TaskTopic); err != nil {
 				log.WithError(err).Error("fire task failed")
 			}
 		}

config.toml

@@ -8,6 +8,8 @@ password="123456"
 db=0
 
 [kafka]
+#brokers="192.168.1.108:9092"
 #brokers="192.168.1.220:9092"
-brokers="192.168.1.108:9092"
-topic="pbaigc"
+brokers="127.0.0.1:9092"
+task_topic="pbaigc"
+receipt_topic="taskreceipt"

config/Config.go

@@ -12,8 +12,9 @@ type RedisConfig struct {
 	DbIndex  int    `json:"db_index" toml:"db_index"`
 }
 type KafkaConfig struct {
-	Brokers string `json:"brokers" toml:"brokers"`
-	Topic   string `json:"topic" toml:"topic"`
+	Brokers      string `json:"brokers" toml:"brokers"`
+	ReceiptTopic string `json:"receipt_topic" toml:"receipt_topic"`
+	TaskTopic    string `json:"task_topic" toml:"task_topic"`
 }
 
 type Config struct {

server/node.go

@@ -12,13 +12,15 @@ import (
 	log "github.com/sirupsen/logrus"
 	"strings"
 	"sync"
+	"time"
 )
 
 type Node struct {
-	rdb  *redis.Client
-	quit chan struct{}
-	conf *config.Config
-	wg   sync.WaitGroup
+	rdb           *redis.Client
+	quit          chan struct{}
+	conf          *config.Config
+	kafkaProducer sarama.AsyncProducer
+	wg            sync.WaitGroup
 }
 
 func NewNode() *Node {
@@ -33,6 +35,8 @@ func NewNode() *Node {
 		quit: make(chan struct{}),
 		conf: config.GetConfig(),
 	}
+	brokers := strings.Split(node.conf.Kafka.Brokers, ";")
+	node.kafkaProducer, _ = utils.NewKafkaProducer(brokers)
 
 	return node
 }
@@ -76,6 +80,29 @@ func (n *Node) Loop(idx int) {
 		}
 		return err
 	}
+	postReceipt := func(task *odysseus.TaskContent, result *odysseus.TaskResponse, err error) error {
+		receipt := new(odysseus.TaskReceipt)
+		receipt.TaskUuid = task.TaskUuid
+		receipt.TaskFinishTime = uint64(time.Now().Unix())
+		receipt.TaskId = task.TaskId
+		receipt.TaskUid = task.TaskUid
+		receipt.TaskWorkload = task.TaskWorkload
+		receipt.TaskDuration = int64(receipt.TaskFinishTime - task.TaskTimestamp)
+		receipt.TaskFee = 0
+		receipt.TaskOutLen = 0
+		receipt.TaskProfitAccount = ""
+		receipt.TaskWorkerAccount = ""
+		switch err {
+		case ErrNoWorker:
+			receipt.TaskResult = err.Error()
+		case ErrDispatchFailed:
+			receipt.TaskResult = err.Error()
+		default:
+			receipt.TaskResult = "internal error"
+		}
+		utils.FireTaskReceipt(n.kafkaProducer, receipt, config.GetConfig().Kafka.ReceiptTopic)
+		return nil
+	}
 
 	for {
 		select {
@@ -96,6 +123,7 @@ func (n *Node) Loop(idx int) {
 						TaskIsSucceed: false,
 						TaskError:     err.Error(),
 					}
+					postReceipt(task, result, err)
 					err = postResult(task, result)
 					if err != nil {
 						log.WithError(err).Error("post task result failed")
@@ -137,7 +165,7 @@ func (n *Node) attachKafkaConsumer(ctx context.Context, taskCh chan *odysseus.Ta
 	}
 
 	consumeFunc := func(consumer *Consumer) {
-		topics := strings.Split(n.conf.Kafka.Topic, ";")
+		topics := strings.Split(n.conf.Kafka.TaskTopic, ";")
 		for {
 			if err := client.Consume(ctx, topics, consumer); err != nil {
 				if errors.Is(err, sarama.ErrClosedConsumerGroup) {

utils/kafka.go

+package utils
+
+import (
+	"github.com/IBM/sarama"
+	"github.com/gogo/protobuf/proto"
+	odysseus "github.com/odysseus/odysseus-protocol/gen/proto/go/base/v1"
+	"time"
+)
+
+// NewKafkaProducer Create a new KafkaProducer.
+func NewKafkaProducer(brokers []string) (sarama.AsyncProducer, error) {
+	kafkaConfig := sarama.NewConfig()
+	kafkaConfig.Producer.RequiredAcks = sarama.WaitForLocal     // Only wait for the leader to ack
+	kafkaConfig.Producer.Compression = sarama.CompressionSnappy // Compress messages
+	kafkaConfig.Producer.Return.Errors = false
+	kafkaConfig.Producer.Flush.Frequency = 500 * time.Millisecond // Flush batches every 500ms
+	kafkaConfig.Net.ResolveCanonicalBootstrapServers = false
+
+	producer, err := sarama.NewAsyncProducer(brokers, kafkaConfig)
+
+	if err != nil {
+		return nil, err
+	}
+
+	go func() {
+		for _ = range producer.Errors() {
+			//log.Printf("Failed to send log entry to kafka: %v\n", err)
+		}
+	}()
+
+	return producer, nil
+}
+
+func FireTaskReceipt(producer sarama.AsyncProducer, receipt *odysseus.TaskReceipt, topic string) error {
+	// Check time for partition key
+	var partitionKey sarama.StringEncoder
+	partitionKey = sarama.StringEncoder(time.Now().Format("2006-01-02"))
+
+	b, err := proto.Marshal(receipt)
+	if err != nil {
+		return err
+	}
+	value := sarama.ByteEncoder(b)
+
+	producer.Input() <- &sarama.ProducerMessage{
+		Key:   partitionKey,
+		Topic: topic,
+		Value: value,
+	}
+
+	return nil
+}