Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

user : tls models exit gracefully. #165

Merged
merged 1 commit into from
Aug 12, 2022
Merged

user : tls models exit gracefully. #165

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 14 additions & 9 deletions cli/cmd/tls.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -109,7 +109,7 @@ func openSSLCommandFunc(command *cobra.Command, args []string) {
conf.SetHex(gConf.IsHex)
conf.SetNoSearch(gConf.NoSearch)

err := conf.Check()
err = conf.Check()

if err != nil {
// ErrorGoBINNotSET is a special error, we should not print it.
Expand All @@ -132,13 +132,14 @@ func openSSLCommandFunc(command *cobra.Command, args []string) {
}

// 加载ebpf，挂载到hook点上，开始监听
go func(module user.IModule) {
err := module.Run()
if err != nil {
logger.Printf("%s\tmodule run failed, [skip it]. error:%+v", module.Name(), err)
return
}
}(mod)
//go func(module user.IModule) {
//
//}(mod)
err = mod.Run()
if err != nil {
logger.Printf("%s\tmodule run failed, [skip it]. error:%+v", mod.Name(), err)
continue
}
runModules[mod.Name()] = mod
logger.Printf("%s\tmodule started successfully.", mod.Name())
wg.Add(1)
Expand All @@ -147,17 +148,21 @@ func openSSLCommandFunc(command *cobra.Command, args []string) {

// needs runmods > 0
if runMods > 0 {
logger.Printf("ECAPTURE :: \tstart to %d modules", runMods)
<-stopper
} else {
logger.Println("ECAPTURE :: \tNo runnable modules, Exit(1)")
os.Exit(1)
}
cancelFun()

// clean up
for _, mod := range runModules {
err = mod.Close()
wg.Done()
if err != nil {
logger.Fatalf("%s\tmodule close failed. error:%+v", mod.Name(), err)
}
wg.Done()
}

wg.Wait()
Expand Down
135 changes: 74 additions & 61 deletions user/imodule.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -132,20 +132,23 @@ func (this *Module) readEvents() error {
for _, event := range this.child.Events() {
switch {
case event.Type() == ebpf.RingBuf:
go this.ringbufEventReader(errChan, event)
this.ringbufEventReader(errChan, event)
case event.Type() == ebpf.PerfEventArray:
go this.perfEventReader(errChan, event)
this.perfEventReader(errChan, event)
default:
errChan <- fmt.Errorf("%s\tNot support mapType:%s , mapinfo:%s", this.child.Name(), event.Type().String(), event.String())
return fmt.Errorf("%s\tNot support mapType:%s , mapinfo:%s", this.child.Name(), event.Type().String(), event.String())
}
}

for {
select {
case err := <-errChan:
return err
go func() {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe simplify this as

go func() {
    for err := <- errChan {
    }
}

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

good suggestion..

for {
select {
case err := <-errChan:
this.logger.Printf("%s\treadEvents error:%v", this.child.Name(), err)
}
}
}
}()
return nil
}

func (this *Module) perfEventReader(errChan chan error, em *ebpf.Map) {
Expand All @@ -154,40 +157,42 @@ func (this *Module) perfEventReader(errChan chan error, em *ebpf.Map) {
errChan <- fmt.Errorf("creating %s reader dns: %s", em.String(), err)
return
}
defer rd.Close()
for {
//判断ctx是不是结束
select {
case _ = <-this.ctx.Done():
this.logger.Printf("%s\tperfEventReader received close signal from context.Done().", this.child.Name())
return
default:
}
this.reader = append(this.reader, rd)
go func() {
for {
//判断ctx是不是结束
select {
case _ = <-this.ctx.Done():
this.logger.Printf("%s\tperfEventReader received close signal from context.Done().", this.child.Name())
return
default:
}

record, err := rd.Read()
if err != nil {
if errors.Is(err, perf.ErrClosed) {
record, err := rd.Read()
if err != nil {
if errors.Is(err, perf.ErrClosed) {
return
}
errChan <- fmt.Errorf("%s\treading from perf event reader: %s", this.child.Name(), err)
return
}
errChan <- fmt.Errorf("%s\treading from perf event reader: %s", this.child.Name(), err)
return
}

if record.LostSamples != 0 {
this.logger.Printf("%s\tperf event ring buffer full, dropped %d samples", this.child.Name(), record.LostSamples)
continue
}
if record.LostSamples != 0 {
this.logger.Printf("%s\tperf event ring buffer full, dropped %d samples", this.child.Name(), record.LostSamples)
continue
}

var event event_processor.IEventStruct
event, err = this.child.Decode(em, record.RawSample)
if err != nil {
this.logger.Printf("%s\tthis.child.decode error:%v", this.child.Name(), err)
continue
}
var event event_processor.IEventStruct
event, err = this.child.Decode(em, record.RawSample)
if err != nil {
this.logger.Printf("%s\tthis.child.decode error:%v", this.child.Name(), err)
continue
}

// 上报数据
this.Dispatcher(event)
}
// 上报数据
this.Dispatcher(event)
}
}()
}

func (this *Module) ringbufEventReader(errChan chan error, em *ebpf.Map) {
Expand All @@ -196,36 +201,38 @@ func (this *Module) ringbufEventReader(errChan chan error, em *ebpf.Map) {
errChan <- fmt.Errorf("%s\tcreating %s reader dns: %s", this.child.Name(), em.String(), err)
return
}
defer rd.Close()
for {
//判断ctx是不是结束
select {
case _ = <-this.ctx.Done():
this.logger.Printf("%s\tringbufEventReader received close signal from context.Done().", this.child.Name())
return
default:
}
this.reader = append(this.reader, rd)
go func() {
for {
//判断ctx是不是结束
select {
case _ = <-this.ctx.Done():
this.logger.Printf("%s\tringbufEventReader received close signal from context.Done().", this.child.Name())
return
default:
}

record, err := rd.Read()
if err != nil {
if errors.Is(err, ringbuf.ErrClosed) {
this.logger.Printf("%s\tReceived signal, exiting..", this.child.Name())
record, err := rd.Read()
if err != nil {
if errors.Is(err, ringbuf.ErrClosed) {
this.logger.Printf("%s\tReceived signal, exiting..", this.child.Name())
return
}
errChan <- fmt.Errorf("%s\treading from ringbuf reader: %s", this.child.Name(), err)
return
}
errChan <- fmt.Errorf("%s\treading from ringbuf reader: %s", this.child.Name(), err)
return
}

var event event_processor.IEventStruct
event, err = this.child.Decode(em, record.RawSample)
if err != nil {
this.logger.Printf("%s\tthis.child.decode error:%v", this.child.Name(), err)
continue
}
var event event_processor.IEventStruct
event, err = this.child.Decode(em, record.RawSample)
if err != nil {
this.logger.Printf("%s\tthis.child.decode error:%v", this.child.Name(), err)
continue
}

// 上报数据
this.Dispatcher(event)
}
// 上报数据
this.Dispatcher(event)
}
}()
}

func (this *Module) Decode(em *ebpf.Map, b []byte) (event event_processor.IEventStruct, err error) {
Expand Down Expand Up @@ -261,6 +268,12 @@ func (this *Module) Dispatcher(event event_processor.IEventStruct) {
}

func (this *Module) Close() error {
this.logger.Printf("%s\tclose", this.child.Name())
for _, iClose := range this.reader {
if err := iClose.Close(); err != nil {
return err
}
}
err := this.processor.Close()
return err
}
2 changes: 1 addition & 1 deletion user/probe_bash.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,7 +73,7 @@ func (this *MBashProbe) Close() error {
if err := this.bpfManager.Stop(manager.CleanAll); err != nil {
return fmt.Errorf("couldn't stop manager %v ", err)
}
return nil
return this.Module.Close()
}

// 通过elf的常量替换方式传递数据
Expand Down
2 changes: 1 addition & 1 deletion user/probe_gnutls.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,7 +77,7 @@ func (this *MGnutlsProbe) Close() error {
if err := this.bpfManager.Stop(manager.CleanAll); err != nil {
return fmt.Errorf("couldn't stop manager %v", err)
}
return nil
return this.Module.Close()
}

// 通过elf的常量替换方式传递数据
Expand Down
2 changes: 1 addition & 1 deletion user/probe_gossl.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -114,5 +114,5 @@ func (p *GoSSLProbe) DecodeFun(m *ebpf.Map) (event_processor.IEventStruct, bool)
}

func (p *GoSSLProbe) Close() error {
return nil
return p.Module.Close()
}
2 changes: 1 addition & 1 deletion user/probe_mysqld.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,7 +84,7 @@ func (this *MMysqldProbe) Close() error {
if err := this.bpfManager.Stop(manager.CleanAll); err != nil {
return fmt.Errorf("couldn't stop manager %v", err)
}
return nil
return this.Module.Close()
}

func (this *MMysqldProbe) setupManagers() error {
Expand Down
2 changes: 1 addition & 1 deletion user/probe_nspr.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,7 +77,7 @@ func (this *MNsprProbe) Close() error {
if err := this.bpfManager.Stop(manager.CleanAll); err != nil {
return fmt.Errorf("couldn't stop manager %v ", err)
}
return nil
return this.Module.Close()
}

// 通过elf的常量替换方式传递数据
Expand Down
2 changes: 1 addition & 1 deletion user/probe_postgres.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,7 +85,7 @@ func (this *MPostgresProbe) Close() error {
if err := this.bpfManager.Stop(manager.CleanAll); err != nil {
return fmt.Errorf("couldn't stop manager %v.", err)
}
return nil
return this.Module.Close()
}

func (this *MPostgresProbe) setupManagers() error {
Expand Down