ebpf/utils/namedpipe/name_pipe_windows.go

//go:build windows
// +build windows

package namedpipe

import (
	"bufio"
	"fmt"
	log "github.com/sirupsen/logrus"
	"net"
	"sync/atomic"
)

const pipeNameDaemonAndSubPattern = `\\.\pipe\`
const ONEAGENT = "cw-oneagent"
const NETFLOW = "cw-netflow"
const SERVERAGENT = "cw-serveragent"

type NamedPipe4W struct {
	*namedPipe
	pipePath string
	listener *npipe.PipeListener
	conn     net.Conn
}

/*
ListenNpipe

	@Description: 创建命名管道文件，返回命名管道对象
	@param  agent_type:   from
	@param  to_type:      to
	@param  isMaster:     是否为daemon创建的
	@param  single:       是否为单文件模式
	@param  rootPath:     命名管道文件夹路径
	@return *NamedPipe4U: 命名管道对象
	@return error:        错误信息
*/
func ListenNpipe(agent_type, to_type string, isMaster, single bool, rootPath string) (*NamedPipe4W, error) {

	np, err := newNamePipe(agent_type, to_type, isMaster, single, rootPath)
	if err != nil {
		return nil, err
	}
	np4w := &NamedPipe4W{namedPipe: np}
	if err := np4w.initPipe(); err != nil {
		return nil, err
	}
	return np4w, nil
}

func (np *NamedPipe4W) getPipePath() (daemon2subPipe, sub2DaemonPipe string, err error) {
	daemon2subPipe = fmt.Sprintf("%s%s2%s", pipeNameDaemonAndSubPattern, np.fromType, np.toType)
	sub2DaemonPipe = fmt.Sprintf("%s%s2%s", pipeNameDaemonAndSubPattern, np.toType, np.fromType)
	return daemon2subPipe, sub2DaemonPipe, nil
}

func (np *NamedPipe4W) initPipe() error {

	daemon2subPipe, sub2DaemonPipe, err := np.getPipePath()

	if err != nil {
		return err
	}

	if np.isDaemon {
		// daemon进程进行建连
		ln, err := npipe.Listen(daemon2subPipe)
		if err != nil {
			return err
		}
		np.listener = ln
		// 过程可能会阻塞
		conn, err := ln.Accept()
		if err != nil {
			return err
		}
		np.conn = conn
		np.recvPipe = bufio.NewReader(conn)
		np.sendPipe = bufio.NewWriter(conn)
	} else {
		conn, err := npipe.Dial(sub2DaemonPipe)
		if err != nil {
			return err
		}
		np.conn = conn
		np.sendPipe = bufio.NewWriter(conn)
		np.recvPipe = bufio.NewReader(conn)
	}
	return nil
}

// only called when io.EOF occurs
func (np *NamedPipe4W) FreeUp4EOF() {
	if !atomic.CompareAndSwapUint32(&np.quitFlag, 0, 1) {
		log.WithField("module", "namedPipe_Win").Warnf("named pipe(%s) already free", np.pipePath)
		return
	}
	log.WithField("module", "namedPipe_Win").Infof("named pipe(%s) free begin", np.pipePath)
	np.wg.Wait()

	conn := np.conn
	np.conn = nil
	if conn != nil {
		if err := conn.Close(); err != nil {
			log.WithField("module", "namedPipe_Win").Errorf("close named pipe(%s) conn error: %v", np.pipePath, err)
		}
	}

	listener := np.listener
	np.listener = nil
	if listener != nil {
		if err := listener.Close(); err != nil {
			log.WithField("module", "namedPipe_Win").Errorf("close named pipe(%s) listener error: %v", np.pipePath, err)
		}
	}

	np.sendPipe = nil
	np.recvPipe = nil
	log.WithField("module", "namedPipe_Win").Infof("named pipe free(%s) end", np.pipePath)
}