-
Notifications
You must be signed in to change notification settings - Fork 2.5k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
create uds server and create base packet server class
- Loading branch information
1 parent
4b45285
commit d86e716
Showing
5 changed files
with
129 additions
and
60 deletions.
There are no files selected for viewing
64 changes: 64 additions & 0 deletions
64
receiver/statsdreceiver/internal/transport/packet_server.go
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,64 @@ | ||
| // Copyright The OpenTelemetry Authors | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
|
|
||
| package transport // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/statsdreceiver/internal/transport" | ||
|
|
||
| import ( | ||
| "errors" | ||
| "net" | ||
|
|
||
| "go.opentelemetry.io/collector/consumer" | ||
| ) | ||
|
|
||
| type packetServer struct { | ||
| packetConn net.PacketConn | ||
| transport Transport | ||
| } | ||
|
|
||
| // ListenAndServe starts the server ready to receive metrics. | ||
| func (u *packetServer) ListenAndServe( | ||
| nextConsumer consumer.Metrics, | ||
| reporter Reporter, | ||
| transferChan chan<- Metric, | ||
| ) error { | ||
| if nextConsumer == nil || reporter == nil { | ||
| return errNilListenAndServeParameters | ||
| } | ||
|
|
||
| buf := make([]byte, 65527) // max size for udp packet body (assuming ipv6) | ||
| for { | ||
| n, addr, err := u.packetConn.ReadFrom(buf) | ||
| if n > 0 { | ||
| u.handlePacket(n, buf, addr, transferChan) | ||
| } | ||
| if err != nil { | ||
| reporter.OnDebugf("%s Transport (%s) - ReadFrom error: %v", | ||
| u.transport, | ||
| u.packetConn.LocalAddr(), | ||
| err) | ||
| var netErr net.Error | ||
| if errors.As(err, &netErr) { | ||
| if netErr.Timeout() { | ||
| continue | ||
| } | ||
| } | ||
| return err | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // handlePacket is helper that parses the buffer and split it line by line to be parsed upstream. | ||
| func (u *packetServer) handlePacket( | ||
| numBytes int, | ||
| data []byte, | ||
| addr net.Addr, | ||
| transferChan chan<- Metric, | ||
| ) { | ||
| splitPacket := NewSplitBytes(data[:numBytes], '\n') | ||
| for splitPacket.Next() { | ||
| chunk := splitPacket.Chunk() | ||
| if len(chunk) > 0 { | ||
| transferChan <- Metric{string(chunk), addr} | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,46 @@ | ||
| // Copyright The OpenTelemetry Authors | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
|
|
||
| package transport // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/statsdreceiver/internal/transport" | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "net" | ||
| "os" | ||
| ) | ||
|
|
||
| type udsServer struct { | ||
| packetServer | ||
| } | ||
|
|
||
| // Ensure that Server is implemented on UDS Server. | ||
| var _ (Server) = (*udsServer)(nil) | ||
|
|
||
| // NewUDSServer creates a transport.Server using Unixgram as its transport. | ||
| func NewUDSServer(transport Transport, socketPath string) (Server, error) { | ||
| if !transport.IsPacketTransport() { | ||
| return nil, fmt.Errorf("NewUDSServer with %s: %w", transport.String(), ErrUnsupportedPacketTransport) | ||
| } | ||
|
|
||
| if _, err := os.Stat(socketPath); err == nil { | ||
| os.Remove(socketPath) | ||
| } | ||
|
|
||
| conn, err := net.ListenPacket(transport.String(), socketPath) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("starting to listen %s socket: %w", transport.String(), err) | ||
| } | ||
|
|
||
| return &udsServer{ | ||
| packetServer: packetServer{ | ||
| packetConn: conn, | ||
| transport: transport, | ||
| }, | ||
| }, nil | ||
| } | ||
|
|
||
| // Close closes the server. | ||
| func (u *udsServer) Close() error { | ||
| os.Remove(u.packetConn.LocalAddr().String()) | ||
| return u.packetConn.Close() | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters