(skywalking-rover) branch reduce-handle-connect-time updated: update conntrack

[liuhan]

This is an automated email from the ASF dual-hosted git repository.

liuhan pushed a commit to branch reduce-handle-connect-time
in repository https://gitbox.apache.org/repos/asf/skywalking-rover.git


The following commit(s) were added to refs/heads/reduce-handle-connect-time by 
this push:
     new fe02798  update conntrack
fe02798 is described below

commit fe027989c882770f46277fd254bf52365a8b5f45
Author: mrproliu <741550...@qq.com>
AuthorDate: Sat Dec 28 15:31:33 2024 +0800

    update conntrack
---
 pkg/tools/ip/conntrack.go | 44 ++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 44 insertions(+)

diff --git a/pkg/tools/ip/conntrack.go b/pkg/tools/ip/conntrack.go
index 978569d..4ccc365 100644
--- a/pkg/tools/ip/conntrack.go
+++ b/pkg/tools/ip/conntrack.go
@@ -58,6 +58,45 @@ func NewConnTrack() (*ConnTrack, error) {
                return nil, err
        }
 
+       go func() {
+               client, _ := conntrack.Dial(nil)
+               evCh := make(chan conntrack.Event, 2048)
+
+               errCh, err := client.Listen(evCh, 4, []netfilter.NetlinkGroup{
+                       netfilter.GroupCTNew, // watching for new conntrack 
events
+               })
+               if err != nil {
+                       log.Error(err)
+               }
+
+               client.SetReadBuffer(26214400) // 25MB
+               // Listen to Conntrack events from all network namespaces on 
the system.
+               err = client.SetOption(netlink.ListenAllNSID, true)
+               if err != nil {
+                       log.Error(err)
+               }
+
+               // Start a goroutine to print all incoming messages on the 
event channel.
+               go func() {
+                       for {
+                               e := <-evCh
+                               if e.Flow.TupleOrig.Proto.DestinationPort == 53 
{
+                                       continue
+                               }
+                               log.Infof("conntrack: type: %s, origin: 
%s:%d->%s:%d, reply: %s:%d->%s:%d", e.Type,
+                                       e.Flow.TupleOrig.IP.SourceAddress, 
e.Flow.TupleOrig.Proto.SourcePort,
+                                       e.Flow.TupleOrig.IP.DestinationAddress, 
e.Flow.TupleOrig.Proto.DestinationPort,
+                                       e.Flow.TupleReply.IP.SourceAddress, 
e.Flow.TupleReply.Proto.SourcePort,
+                                       
e.Flow.TupleReply.IP.DestinationAddress, 
e.Flow.TupleReply.Proto.DestinationPort)
+                       }
+               }()
+
+               // Stop the program as soon as an error is caught in a decoder 
goroutine.
+               if err := <-errCh; err != nil {
+                       log.Errorf("conntrack error: %v", err)
+               }
+       }()
+
        return &ConnTrack{
                queryClient:   queryClient,
                eventChain:    make(chan conntrack.Event, 2048),
@@ -138,6 +177,11 @@ func (c *ConnTrack) monitor0(ctx context.Context) (chan 
error, error) {
                        case <-ctx.Done():
                                return
                        case e := <-c.eventChain:
+                               log.Infof("----------conntrack: type: %s, 
origin: %s:%d->%s:%d, reply: %s:%d->%s:%d", e.Type,
+                                       e.Flow.TupleOrig.IP.SourceAddress, 
e.Flow.TupleOrig.Proto.SourcePort,
+                                       e.Flow.TupleOrig.IP.DestinationAddress, 
e.Flow.TupleOrig.Proto.DestinationPort,
+                                       e.Flow.TupleReply.IP.SourceAddress, 
e.Flow.TupleReply.Proto.SourcePort,
+                                       
e.Flow.TupleReply.IP.DestinationAddress, 
e.Flow.TupleReply.Proto.DestinationPort)
                                if e.Flow.TupleOrig.Proto.DestinationPort == 53 
{
                                        continue
                                }