How to send vector proto messages via rpc ?

[evanzhang87]

Hello, I want to send messages (log messages) to vector source with my code, so I tried to use grpc, but it didn't work.
I use event.proto and vector.proto to generate code.

func TestVector(t *testing.T) {
	wrapper := &event_pb.EventWrapper{
		Event: &event_pb.EventWrapper_Log{
			Log: &event_pb.Log{
				Value: &event_pb.Value{
					Kind: &event_pb.Value_RawBytes{
						RawBytes: []byte("hello world"),
					},
				},
			},
		},
	}
	var events []*event_pb.EventWrapper
	events = append(events, wrapper)
	req := vector_pb.PushEventsRequest{
		Events: events,
	}
	addr := "xx.xx.xx.xx:9000"
	conn, err := grpc.NewClient(addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		log.Fatal(err)
	}
	cli := vector_pb.NewVectorClient(conn)

	ctx := context.TODO()
	t.Logf(req.String())
	res, err := cli.PushEvents(ctx, &req)
	if err != nil {
		log.Fatal(err)
	}
	t.Logf(res.String())
}

// test log

=== RUN   TestVector
    vector_test.go:38: events:{log:{value:{raw_bytes:"hello world"}}}
    vector_test.go:43: 
--- PASS: TestVector (0.01s)
PASS

res.String() always be "", is that OK?

// vector config
sources:
  vector:
    type: vector
    address: 0.0.0.0:9000
sinks:
  file:
    type: file
    inputs:
      - vector
    path: /var/log/vector/output/vector.log
    encoding:
      codec: json

// vector source prints
{"source_type":"vector","timestamp":"2024-09-10T11:46:43.607741476Z"}
{"source_type":"vector","timestamp":"2024-09-10T11:48:37.998557858Z"}
{"source_type":"vector","timestamp":"2024-09-10T11:51:32.877463923Z"}

I can't get raw_bytes.

[evanzhang87]

Wow, it works when I send a value map.

	fields := make(map[string]*event_pb.Value)
	fields["message"] = &event_pb.Value{
		Kind: &event_pb.Value_RawBytes{
			RawBytes: []byte("hello world"),
		},
	}
	fields["time"] = &event_pb.Value{
		Kind: &event_pb.Value_Timestamp{
			Timestamp: timestamppb.Now(),
		},
	}

	wrapper := &event_pb.EventWrapper{
		Event: &event_pb.EventWrapper_Log{
			Log: &event_pb.Log{
				Value: &event_pb.Value{
					Kind: &event_pb.Value_Map{
						Map: &event_pb.ValueMap{
							Fields: fields,
						},
					},
				},
			},
		},
	}

[evanzhang87]

I'm trying to replace logstash with vector, but when vector establishes too many connections (> 1k), it's response will be slower. #20969
My bench:

// It's a simple config, logstash1 will accept online logs, logstash2 just accept test logs.
source:
 - logstash1
 - logstash2
transforms:
 transforms1
  - logstash1
  - logstash2
sink:
 kafka:
  - transforms1

Then I run 2 filebeat to collect the same log file but send to different logstash source , logstash2 looks faster, and all _utils are below 0.5.

So I think the logstash source is a bottleneck when establishes too many connections.
Therefore, I want to build a gateway to accept logs of logstash connections and then sent to vector via vector source.
Is there any suggestions?