func closeSniffers(sniffs []sniffer) results {
	requestClose(sniffs)
	r := gatherResponse(sniffs)
	closeHandles(sniffs)
	return r
}
Ask each sniffer to stop collection, gather the total
statistics, and then attempt to close pcap handles just
to be safe (should be closed by requestClose())
package main
import (
	"errors"
	"flag"
	"fmt"
	"io"
	"net"
	"os"
	"path"
	"runtime/pprof"
	"syscall"
	"time"
	"github.com/gravwell/ingest/v3"
	"github.com/gravwell/ingest/v3/entry"
	"github.com/gravwell/ingest/v3/log"
	"github.com/gravwell/ingesters/v3/utils"
	"github.com/gravwell/ingesters/v3/version"
	"github.com/google/gopacket/layers"
	"github.com/google/gopacket/pcap"
)
const (
	defaultConfigLoc     = `/opt/gravwell/etc/network_capture.conf`
	packetsThrowSize int = 1024 * 1024 * 2
)
var (
	confLoc        = flag.String("config-file", defaultConfigLoc, "Location for configuration file")
	verbose        = flag.Bool("v", false, "Display verbose status updates to stdout")
	stderrOverride = flag.String("stderr", "", "Redirect stderr to a shared memory file")
	profileFile    = flag.String("profile", "", "Start a CPU profiler, disabled if blank")
	ver            = flag.Bool("version", false, "Print the version information and exit")
	pktTimeout time.Duration = 500 * time.Millisecond
	totalPackets uint64
	totalBytes   uint64
	v            bool
	lg           *log.Logger
)
type results struct {
	Bytes uint64
	Count uint64
	Error error
}
type sniffer struct {
	name      string
	Promisc   bool
	Interface string
	TagName   string
	tag       entry.EntryTag
	SnapLen   int
	BPFFilter string
	handle    *pcap.Handle
	src       net.IP
	die       chan bool
	res       chan results
	active    bool
}
func init() {
	flag.Parse()
	if *ver {
		version.PrintVersion(os.Stdout)
		ingest.PrintVersion(os.Stdout)
		os.Exit(0)
	}
	lg = log.New(os.Stderr)  
	if *stderrOverride != `` {
		if oldstderr, err := syscall.Dup(int(os.Stderr.Fd())); err != nil {
			lg.Fatal("Failed to dup stderr: %v\n", err)
		} else {
			lg.AddWriter(os.NewFile(uintptr(oldstderr), "oldstderr"))
		}
		fp := path.Join(`/dev/shm/`, *stderrOverride)
		fout, err := os.Create(fp)
		if err != nil {
			fmt.Fprintf(os.Stderr, "Failed to create %s: %v\n", fp, err)
		} else {
			version.PrintVersion(fout)
			ingest.PrintVersion(fout)
			if err := syscall.Dup2(int(fout.Fd()), int(os.Stderr.Fd())); err != nil {
				fmt.Fprintf(os.Stderr, "Failed to dup2 stderr: %v\n", err)
				fout.Close()
			}
		}
	}
	v = *verbose
}
func main() {
	if *profileFile != `` {
		f, err := os.Create(*profileFile)
		if err != nil {
			lg.Fatal("failed to open pprof: %v", err)
		}
		defer f.Close()
		pprof.StartCPUProfile(f)
		defer pprof.StopCPUProfile()
	}
	cfg, err := GetConfig(*confLoc)
	if err != nil {
		lg.Fatal("Failed to get configuration: %v", err)
	}
	if len(cfg.Log_File) > 0 {
		fout, err := os.OpenFile(cfg.Log_File, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0640)
		if err != nil {
			lg.FatalCode(0, "Failed to open log file %s: %v", cfg.Log_File, err)
		}
		if err = lg.AddWriter(fout); err != nil {
			lg.Fatal("Failed to add a writer: %v", err)
		}
		if len(cfg.Log_Level) > 0 {
			if err = lg.SetLevelString(cfg.Log_Level); err != nil {
				lg.FatalCode(0, "Invalid Log Level \"%s\": %v", cfg.Log_Level, err)
			}
		}
	}
	tags, err := cfg.Tags()
	if err != nil {
		lg.FatalCode(0, "Failed to get tags from configuration: %v", err)
	}
	conns, err := cfg.Targets()
	if err != nil {
		lg.FatalCode(0, "Failed to get backend targets from configuration: %v", err)
	}
	debugout("Handling %d tags over %d targets\n", len(tags), len(conns))
	debugout("INSECURE skipping TLS verification: %v\n", cfg.InsecureSkipTLSVerification())
	id, ok := cfg.IngesterUUID()
	if !ok {
		lg.FatalCode(0, "Couldn't read ingester UUID\n")
	}
	igCfg := ingest.UniformMuxerConfig{
		Destinations:    conns,
		Tags:            tags,
		Auth:            cfg.Secret(),
		LogLevel:        cfg.LogLevel(),
		IngesterName:    "networkLog",
		IngesterVersion: version.GetVersion(),
		IngesterUUID:    id.String(),
		VerifyCert:      !cfg.InsecureSkipTLSVerification(),
		Logger:          lg,
	}
	if cfg.EnableCache() {
		igCfg.EnableCache = true
		igCfg.CacheConfig.FileBackingLocation = cfg.LocalFileCachePath()
		igCfg.CacheConfig.MaxCacheSize = cfg.MaxCachedData()
	}
	igst, err := ingest.NewUniformMuxer(igCfg)
	if err != nil {
		lg.Fatal("Failed to create new uniform muxer: %v ", err)
	}
	debugout("Started ingester muxer\n")
	if err := igst.Start(); err != nil {
		lg.Fatal("Failed start our ingest system: %v", err)
	}
	debugout("Waiting for connections to indexers\n")
	if err := igst.WaitForHot(cfg.Timeout()); err != nil {
		lg.Fatal("Timedout waiting for backend connections: %v", err)
	}
	debugout("Successfully connected to ingesters\n")
	var sniffs []sniffer
	for k, v := range cfg.Sniffer {
		if v == nil {
			closeSniffers(sniffs)
			lg.FatalCode(0, "Invalid sniffer named %s: Nil struct", k)
		}
		var src net.IP
		if v.Source_Override != `` {
			src = net.ParseIP(v.Source_Override)
			if src == nil {
				closeSniffers(sniffs)
				lg.FatalCode(0, "Source-Override is invalid")
			}
		} else if cfg.Source_Override != `` {
			src = net.ParseIP(cfg.Source_Override)
			if src == nil {
				closeSniffers(sniffs)
				lg.FatalCode(0, "Global Source-Override is invalid")
			}
		} else {
			src, err = getSourceIP(v.Interface)
			if err != nil {
				if src, err = igst.SourceIP(); err != nil {
					closeSniffers(sniffs)
					lg.FatalCode(0, "Failed to get source for %s: %v", v.Interface, err)
				}
			}
		}
		hnd, err := pcap.OpenLive(v.Interface, int32(v.Snap_Len), v.Promisc, pktTimeout)
		if err != nil {
			closeSniffers(sniffs)
			lg.FatalCode(0, "Failed to get initialize handler on %s for %s", v.Interface, k)
		}
		if v.BPF_Filter != `` {
			if err := hnd.SetBPFFilter(v.BPF_Filter); err != nil {
				hnd.Close()
				closeSniffers(sniffs)
				lg.FatalCode(0, "Invalid BPF Filter for %s: %v", k, err)
			}
		}
		sniffs = append(sniffs, sniffer{
			name:      k,
			src:       src,
			Promisc:   v.Promisc,
			Interface: v.Interface,
			TagName:   v.Tag_Name,
			SnapLen:   v.Snap_Len,
			BPFFilter: v.BPF_Filter,
			handle:    hnd,
			die:       make(chan bool, 1),
			res:       make(chan results, 1),
		})
	}
	for i := range sniffs {
		tag, err := igst.GetTag(sniffs[i].TagName)
		if err != nil {
			closeSniffers(sniffs)
			lg.Fatal("Failed to resolve tag %s: %v", sniffs[i].TagName, err)
		}
		sniffs[i].tag = tag
	}
	start := time.Now()
	for i := range sniffs {
		sniffs[i].active = true
		go pcapIngester(igst, &sniffs[i])
	}
	utils.WaitForQuit()
	requestClose(sniffs)
	res := gatherResponse(sniffs)
	closeHandles(sniffs)
	if err := igst.Sync(time.Second); err != nil {
		lg.Error("Failed to sync the ingester: %v\n", err)
	}
	durr := time.Since(start)
	if err == nil {
		lg.Info("Completed in %v (%s)\n", durr, ingest.HumanSize(res.Bytes))
		lg.Info("Total Count: %s\n", ingest.HumanCount(res.Count))
		lg.Info("Entry Rate: %s\n", ingest.HumanEntryRate(res.Count, durr))
		lg.Info("Ingest Rate: %s\n", ingest.HumanRate(res.Bytes, durr))
	}
	if err := igst.Close(); err != nil {
		lg.Error("Failed to close the ingester: %v\n", err)
		return
	}
}
func rebuildPacketSource(s *sniffer) (*pcap.Handle, bool) {
	var threwErr bool
mainLoop:
	for {
		select {
		case <-time.After(time.Second):
		case <-s.die:
			break mainLoop
		}
		hnd, err := pcap.OpenLive(s.Interface, int32(s.SnapLen), s.Promisc, pktTimeout)
		if err != nil {
			if !threwErr {
				threwErr = true
				lg.Error("Failed to get pcap device on reopen (%v)\n", err)
			}
			continue
		}
		if s.BPFFilter != `` {
			if err := hnd.SetBPFFilter(s.BPFFilter); err != nil {
				lg.Error("Invalid BPF Filter on reopen: %v\n", err)
				hnd.Close()
				break mainLoop
			}
		}
		return hnd, true
	}
	return nil, false  
}
type capPacket struct {
	ts   entry.Timestamp
	data []byte
}
func packetExtractor(hnd *pcap.Handle, c chan []capPacket) {
	defer close(c)
	var packets []capPacket
	var packetsSize int
	var capPkt capPacket
	tckr := time.NewTicker(time.Second)
	defer tckr.Stop()
	var trimSize int
	if hnd.LinkType() == layers.LinkTypeLinuxSLL {
		trimSize = 2
	}
	for {
		data, ci, err := hnd.ReadPacketData()
		if err != nil {
			if err == pcap.NextErrorTimeoutExpired || err == io.EOF {
				if len(packets) > 0 {
					c <- packets
					packets = nil
					packetsSize = 0
				}
				continue
			}
			debugout("Failed to get packet from source: %v\n", err)
			break
		}
		if trimSize > 0 && len(data) > trimSize {
			data = data[trimSize:]
		}
		capPkt.data = data
		capPkt.ts = entry.FromStandard(ci.Timestamp)
		packets = append(packets, capPkt)
		packetsSize += len(capPkt.data)
		select {
		case _ = <-tckr.C:
			if len(packets) > 0 {
				c <- packets
				packets = nil
				packetsSize = 0
			}
		default:
			if packetsSize >= packetsThrowSize {
				c <- packets
				packets = nil
				packetsSize = 0
			}
		}
	}
	if len(packets) > 0 {
		c <- packets
	}
}
func pcapIngester(igst *ingest.IngestMuxer, s *sniffer) {
	count := uint64(0)
	totalBytes := uint64(0)
	ch := make(chan []capPacket, 1024)
	go packetExtractor(s.handle, ch)
	debugout("Starting sniffer %s on %s with \"%s\"\n", s.name, s.Interface, s.BPFFilter)
	igst.Info("Starting sniffer %s on %s with \"%s\"\n", s.name, s.Interface, s.BPFFilter)
	lg.Info("Starting sniffer %s on %s with \"%s\"\n", s.name, s.Interface, s.BPFFilter)
mainLoop:
	for {
		select {
		case _ = <-s.die:
			s.handle.Close()
			break mainLoop
		case pkts, ok := <-ch:  
			if !ok {
				igst.Error("Failed to read next packet, attempting to rebuild packet source")
				s.handle.Close()
				s.handle, ok = rebuildPacketSource(s)
				if !ok {
					igst.Error("Couldn't rebuild packet start")
					break mainLoop
				}
				ch = make(chan []capPacket, 1024)
				go packetExtractor(s.handle, ch)
				debugout("Rebuilding packet source\n")
				igst.Info("Rebuilt packet source")
				continue
			}
			staticSet := make([]entry.Entry, len(pkts))
			set := make([]*entry.Entry, len(pkts))
			for i := range pkts {
				staticSet[i].TS = pkts[i].ts
				staticSet[i].Data = pkts[i].data
				staticSet[i].SRC = s.src
				staticSet[i].Tag = s.tag
				set[i] = &staticSet[i]
				totalBytes += uint64(len(pkts[i].data))
				count++
			}
			if err := igst.WriteBatch(set); err != nil {
				s.handle.Close()
				lg.Error("Failed to write entry: %v\n", err)
				s.res <- results{
					Bytes: 0,
					Count: 0,
					Error: err,
				}
				return
			}
		}
	}
	s.res <- results{
		Bytes: totalBytes,
		Count: count,
		Error: nil,
	}
}
func getSourceIP(dev string) (net.IP, error) {
	iface, err := net.InterfaceByName(dev)
	if err != nil {
		return nil, err
	}
	addrs, err := iface.Addrs()
	if err != nil {
		return nil, err
	}
	for _, addr := range addrs {
		ip, _, err := net.ParseCIDR(addr.String())
		if err == nil {
			return ip, nil
		}
		ip = net.ParseIP(addr.String())
		if ip != nil {
			return ip, nil
		}
	}
	return nil, errors.New("No IP for " + dev)
}
func debugout(format string, args ...interface{}) {
	if !v {
		return
	}
	fmt.Printf(format, args...)
}
func addResults(dst *results, src results) {
	if dst == nil {
		return
	}
	dst.Bytes += src.Bytes
	dst.Count += src.Count
}
func requestClose(sniffs []sniffer) {
	for _, s := range sniffs {
		if s.active {
			s.die <- true
		}
	}
}
func gatherResponse(sniffs []sniffer) results {
	var r results
	for _, s := range sniffs {
		if s.active {
			addResults(&r, <-s.res)
		}
	}
	return r
}
func closeHandles(sniffs []sniffer) {
	for _, s := range sniffs {
		if s.handle != nil {
			s.handle.Close()
		}
	}
}