in plugins/forwarder/grpc/nativelog/forwarder.go [74:101]
func (f *Forwarder) Forward(batch event.BatchEvents) error {
for _, e := range batch {
data, ok := e.GetData().(*v1.SniffData_LogList)
if !ok {
continue
}
stream, err := f.logClient.Collect(context.Background())
if err != nil {
log.Logger.Errorf("open grpc stream error %v", err)
return err
}
streamClosed := false
for _, logData := range data.LogList.Logs {
err := stream.SendMsg(server_grpc.NewOriginalData(logData))
if err != nil {
log.Logger.Errorf("%s send log data error: %v", f.Name(), err)
f.closeStream(stream)
streamClosed = true
break
}
}
if !streamClosed {
f.closeStream(stream)
}
}
return nil
}