func NewSaver(host string, pod string, numMarshaller int, srv eventsocket.Server, anon anonymize.IPAnonymizer) *Saver {
	m := make([]MarshalChan, 0, numMarshaller)
	c := cache.NewCache()
	conn := make(map[uint64]*Connection, 500)
	wg := &sync.WaitGroup{}
	wg.Add(1)
	ageLim := 10 * time.Minute
	for i := 0; i < numMarshaller; i++ {
		m = append(m, newMarshaller(wg, anon))
	}
	return &Saver{
		Host:         host,
		Pod:          pod,
		FileAgeLimit: ageLim,
		MarshalChans: m,
		Done:         wg,
		Connections:  conn,
		ClosingStats: make(map[uint64]TcpStats, 100),
		cache:        c,
		eventServer:  srv,
	}
}
NewSaver creates a new Saver for the given host and pod.  numMarshaller controls
how many marshalling goroutines are used to distribute the marshalling workload.
package saver
import (
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"os"
	"sync"
	"sync/atomic"
	"time"
	"github.com/m-lab/go/anonymize"
	"github.com/m-lab/tcp-info/cache"
	"github.com/m-lab/tcp-info/eventsocket"
	"github.com/m-lab/tcp-info/inetdiag"
	"github.com/m-lab/tcp-info/metrics"
	"github.com/m-lab/tcp-info/netlink"
	"github.com/m-lab/tcp-info/tcp"
	"github.com/m-lab/tcp-info/zstd"
	"github.com/m-lab/uuid"
)
const maxSwitchSpeed = 1e10
var (
	ErrNoMarshallers = errors.New("Saver has zero Marshallers")
)
type Task struct {
	Message *netlink.ArchivalRecord
	Writer  io.WriteCloser
}
type CacheLogger interface {
	LogCacheStats(localCount, errCount int)
}
type MarshalChan chan<- Task
func runMarshaller(taskChan <-chan Task, wg *sync.WaitGroup, anon anonymize.IPAnonymizer) {
	for task := range taskChan {
		if task.Message == nil {
			task.Writer.Close()
			continue
		}
		if task.Writer == nil {
			log.Fatal("Nil writer")
		}
		err := task.Message.RawIDM.Anonymize(anon)
		if err != nil {
			log.Println("Failed to anonymize message:", err)
			continue
		}
		b, _ := json.Marshal(task.Message)  
		task.Writer.Write(b)
		task.Writer.Write([]byte("\n"))
	}
	log.Println("Marshaller Done")
	wg.Done()
}
func newMarshaller(wg *sync.WaitGroup, anon anonymize.IPAnonymizer) MarshalChan {
	marshChan := make(chan Task, 100)
	wg.Add(1)
	go runMarshaller(marshChan, wg, anon)
	return marshChan
}
type Connection struct {
	Inode      uint32  
	ID         inetdiag.SockID
	UID        uint32
	Slice      string     
	StartTime  time.Time  
	Sequence   int        
	Expiration time.Time  
	Writer     io.WriteCloser
}
func newConnection(info *inetdiag.InetDiagMsg, timestamp time.Time) *Connection {
	conn := Connection{Inode: info.IDiagInode, ID: info.ID.GetSockID(), UID: info.IDiagUID, Slice: "", StartTime: timestamp, Sequence: 0,
		Expiration: time.Now()}
	return &conn
}
func (conn *Connection) Rotate(Host string, Pod string, FileAgeLimit time.Duration) error {
	datePath := conn.StartTime.Format("2006/01/02")
	if conn.Sequence > 0 {
		now := time.Now().UTC()
		datePath = now.Format("2006/01/02")
	}
	err := os.MkdirAll(datePath, 0777)
	if err != nil {
		return err
	}
	id := uuid.FromCookie(conn.ID.CookieUint64())
	conn.Writer, err = zstd.NewWriter(fmt.Sprintf("%s/%s.%05d.jsonl.zst", datePath, id, conn.Sequence))
	if err != nil {
		return err
	}
	conn.writeHeader()
	metrics.NewFileCount.Inc()
	conn.Expiration = conn.Expiration.Add(10 * time.Minute)
	conn.Sequence++
	return nil
}
func (conn *Connection) writeHeader() {
	msg := netlink.ArchivalRecord{
		Metadata: &netlink.Metadata{
			UUID:      uuid.FromCookie(conn.ID.CookieUint64()),
			Sequence:  conn.Sequence,
			StartTime: conn.StartTime,
		},
	}
	bytes, _ := json.Marshal(msg)
	conn.Writer.Write(bytes)
	conn.Writer.Write([]byte("\n"))
}
type stats struct {
	TotalCount   int64
	NewCount     int64
	DiffCount    int64
	ExpiredCount int64
}
func (s *stats) IncTotalCount() {
	atomic.AddInt64(&s.TotalCount, 1)
}
func (s *stats) IncNewCount() {
	atomic.AddInt64(&s.NewCount, 1)
}
func (s *stats) IncDiffCount() {
	atomic.AddInt64(&s.DiffCount, 1)
}
func (s *stats) IncExpiredCount() {
	atomic.AddInt64(&s.ExpiredCount, 1)
}
func (s *stats) Copy() stats {
	result := stats{}
	result.TotalCount = atomic.LoadInt64(&s.TotalCount)
	result.NewCount = atomic.LoadInt64(&s.NewCount)
	result.DiffCount = atomic.LoadInt64(&s.DiffCount)
	result.ExpiredCount = atomic.LoadInt64(&s.ExpiredCount)
	return result
}
type TcpStats struct {
	Sent     uint64  
	Received uint64  
}
type Saver struct {
	Host          string  
	Pod           string  
	FileAgeLimit  time.Duration
	MarshalChans  []MarshalChan
	Done          *sync.WaitGroup  
	Connections   map[uint64]*Connection
	ClosingStats  map[uint64]TcpStats  
	ClosingTotals TcpStats
	cache       *cache.Cache
	stats       stats
	eventServer eventsocket.Server
}