Skip to content

Commit b23d6db

Browse files
randomswdevrandomswdev
committed
Initial changes
1 parent 29d204a commit b23d6db

File tree

5 files changed

+252
-116
lines changed

5 files changed

+252
-116
lines changed

‎.vscode/launch.json

+17
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,17 @@
1+
{
2+
// Use IntelliSense to learn about possible attributes.
3+
// Hover to view descriptions of existing attributes.
4+
// For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
5+
"version": "0.2.0",
6+
"configurations": [
7+
{
8+
"name": "Launch",
9+
"type": "go",
10+
"request": "launch",
11+
"mode": "auto",
12+
"program": "arachned/main.go",
13+
"env": {},
14+
"args": []
15+
}
16+
]
17+
}

‎bootstrap.go

+56-102
Original file line numberDiff line numberDiff line change
@@ -23,16 +23,13 @@ package arachne
2323
import (
2424
coreLog "log"
2525
"os"
26-
"sync"
2726
"time"
2827

29-
"github.com/uber/arachne/collector"
3028
"github.com/uber/arachne/config"
3129
d "github.com/uber/arachne/defines"
32-
"github.com/uber/arachne/internal/ip"
3330
"github.com/uber/arachne/internal/log"
34-
"github.com/uber/arachne/internal/tcp"
3531
"github.com/uber/arachne/internal/util"
32+
"github.com/uber/arachne/pkg/engine"
3633

3734
"go.uber.org/zap"
3835
)
@@ -87,115 +84,72 @@ func Run(ec *config.Extended, opts ...Option) {
8784
logger.Error("error initializing stats", zap.Error(err))
8885
}
8986

90-
// Hold raw socket connection for IPv4 packets
91-
var connIPv4 *ip.Conn
87+
engineCallbacks := newEngineCallbacks(&gl, logger)
88+
89+
engine := engine.NewEngine(&gl.Engine, logger, engineCallbacks, sr, engineModeFromCLIConfig(gl.CLI), sigC)
9290

9391
logger.Info("Starting up arachne")
9492

95-
for {
96-
var (
97-
err error
98-
currentDSCP ip.DSCPValue
99-
dnsWg sync.WaitGroup
100-
finishedCycleUpload sync.WaitGroup
101-
)
102-
103-
// Channels to tell goroutines to terminate
104-
killC := new(util.KillChannels)
105-
106-
// If Orchestrator mode enabled, fetch JSON configuration file, otherwise try
107-
// to retrieve default local file
108-
err = config.FetchRemoteList(&gl, d.MaxNumRemoteTargets, d.MaxNumSrcTCPPorts,
109-
d.MinBatchInterval, d.HTTPResponseHeaderTimeout, d.OrchestratorRESTConf, sigC, logger)
110-
if err != nil {
111-
break
112-
}
113-
logger.Debug("Global JSON configuration", zap.Any("configuration", gl.RemoteConfig))
93+
engine.Run()
11494

115-
if len(gl.Remotes) == 0 {
116-
logger.Debug("No targets to be echoed have been specified")
117-
apply(&gl, ReceiverOnlyMode(true))
118-
}
95+
// Clean-up
96+
sr.Close()
97+
util.RemovePID(gl.App.PIDPath, logger)
11998

120-
configRefresh := time.NewTicker(gl.RemoteConfig.PollOrchestratorInterval.Success)
121-
122-
if gl.RemoteConfig.ResolveDNS && !*gl.CLI.ReceiverOnlyMode {
123-
// Refresh DNS resolutions
124-
dnsRefresh := time.NewTicker(d.DNSRefreshInterval)
125-
dnsWg.Add(1)
126-
killC.DNSRefresh = make(chan struct{})
127-
config.ResolveDNSTargets(gl.Remotes, gl.RemoteConfig, dnsRefresh, &dnsWg,
128-
killC.DNSRefresh, logger)
129-
dnsWg.Wait()
130-
logger.Debug("Remotes after DNS resolution include",
131-
zap.Int("count", len(gl.Remotes)),
132-
zap.Any("remotes", gl.Remotes))
133-
}
99+
logger.Info("Exiting arachne")
134100

135-
// Channels for Collector to receive Probes and Responses from.
136-
sentC := make(chan tcp.Message, d.ChannelOutBufferSize)
137-
rcvdC := make(chan tcp.Message, d.ChannelInBufferSize)
138-
139-
// Connection for IPv4 packets
140-
if connIPv4 == nil {
141-
connIPv4 = ip.NewConn(
142-
d.AfInet,
143-
gl.RemoteConfig.TargetTCPPort,
144-
gl.RemoteConfig.InterfaceName,
145-
gl.RemoteConfig.SrcAddress,
146-
logger)
147-
}
101+
os.Exit(0)
102+
}
148103

149-
// Actual echoing is a percentage of the total configured batch cycle duration.
150-
realBatchInterval := time.Duration(float32(gl.RemoteConfig.BatchInterval) *
151-
d.BatchIntervalEchoingPerc)
152-
uploadBatchInterval := time.Duration(float32(gl.RemoteConfig.BatchInterval) *
153-
d.BatchIntervalUploadStats)
154-
batchEndCycle := time.NewTicker(uploadBatchInterval)
155-
completeCycleUpload := make(chan bool, 1)
156-
157-
if !*gl.CLI.SenderOnlyMode && !*gl.CLI.ReceiverOnlyMode {
158-
// Start gathering and reporting results.
159-
killC.Collector = make(chan struct{})
160-
collector.Run(&gl, sentC, rcvdC, gl.Remotes, &currentDSCP, sr, completeCycleUpload,
161-
&finishedCycleUpload, killC.Collector, logger)
162-
}
104+
type engineCallbacks struct {
105+
configRefreshTicker *time.Ticker
106+
gl *config.Global
107+
logger *log.Logger
108+
}
163109

164-
if !*gl.CLI.SenderOnlyMode {
165-
// Listen for responses or probes from other IPv4 arachne agents.
166-
killC.Receiver = make(chan struct{})
167-
err = tcp.Receiver(connIPv4, sentC, rcvdC, killC.Receiver, logger)
168-
if err != nil {
169-
logger.Fatal("IPv4 receiver failed to start", zap.Error(err))
170-
}
171-
logger.Debug("IPv4 receiver now ready...")
172-
//TODO IPv6 receiver
173-
}
110+
func newEngineCallbacks(gl *config.Global, logger *log.Logger) *engineCallbacks {
111+
return &engineCallbacks{
112+
gl: gl,
113+
logger: logger,
114+
}
115+
}
174116

175-
if !*gl.CLI.ReceiverOnlyMode {
176-
logger.Debug("Echoing...")
177-
// Start echoing all targets.
178-
killC.Echo = make(chan struct{})
179-
tcp.EchoTargets(gl.Remotes, connIPv4, gl.RemoteConfig.TargetTCPPort,
180-
gl.RemoteConfig.SrcTCPPortRange, gl.RemoteConfig.QoSEnabled, &currentDSCP,
181-
realBatchInterval, batchEndCycle, sentC, *gl.CLI.SenderOnlyMode,
182-
completeCycleUpload, &finishedCycleUpload, killC.Echo, logger)
183-
}
117+
func (ec *engineCallbacks) FetchRemoteList(_ *config.Engine, stopChannel <-chan struct{}, logger *log.Logger) (<-chan struct{}, error) {
118+
err := config.FetchRemoteList(ec.gl, d.MaxNumRemoteTargets, d.MaxNumSrcTCPPorts,
119+
d.MinBatchInterval, d.HTTPResponseHeaderTimeout, d.OrchestratorRESTConf, stopChannel, ec.logger)
120+
if err != nil {
121+
return nil, err
122+
}
123+
124+
ec.configRefreshTicker = time.NewTicker(ec.gl.RemoteConfig.PollOrchestratorInterval.Success)
184125

126+
ch := make(chan struct{})
127+
128+
go func() {
185129
select {
186-
case <-configRefresh.C:
187-
util.CleanUpRefresh(killC, *gl.CLI.ReceiverOnlyMode,
188-
*gl.CLI.SenderOnlyMode, gl.RemoteConfig.ResolveDNS)
189-
log.ResetLogFiles(gl.App.Logging.OutputPaths, d.LogFileSizeMaxMB, d.LogFileSizeKeepKB, logger)
190-
logger.Info("Refreshing target list file, if needed")
191-
configRefresh.Stop()
192-
case <-sigC:
193-
logger.Debug("Received SIG")
194-
configRefresh.Stop()
195-
util.CleanUpAll(killC, *gl.CLI.ReceiverOnlyMode, *gl.CLI.SenderOnlyMode,
196-
gl.RemoteConfig.ResolveDNS, connIPv4, gl.App.PIDPath, sr, logger)
197-
logger.Info("Exiting")
198-
os.Exit(0)
130+
case <-ec.configRefreshTicker.C:
131+
ch <- struct{}{}
132+
}
133+
}()
134+
135+
return ch, err
136+
}
137+
138+
func (ec *engineCallbacks) FetchRemoteListNeeded() {
139+
ec.configRefreshTicker.Stop()
140+
log.ResetLogFiles(ec.gl.App.Logging.OutputPaths, d.LogFileSizeMaxMB, d.LogFileSizeKeepKB, ec.logger)
141+
}
142+
143+
func (ec *engineCallbacks) Stopping() {
144+
ec.configRefreshTicker.Stop()
145+
}
146+
147+
func engineModeFromCLIConfig(conf *config.CLIConfig) engine.EngineMode {
148+
if *conf.ReceiverOnlyMode {
149+
if *conf.SenderOnlyMode {
150+
return engine.EngineSendReceiveMode
199151
}
152+
return engine.EngineReceiveOnlyMode
200153
}
154+
return engine.EngineSendOnlyMode
201155
}

‎collector/collector.go

+5-5
Original file line numberDiff line numberDiff line change
@@ -160,7 +160,7 @@ func (rs resultStore) walkResults(
160160

161161
// processResults calculates metrics, uploads stats and stores in results[] for stdout, if needed.
162162
func (rs resultStore) processResults(
163-
gl *config.Global,
163+
gl *config.Engine,
164164
remotes config.RemoteStore,
165165
target string,
166166
req tcp.Message,
@@ -195,7 +195,7 @@ func (rs resultStore) processResults(
195195
}
196196

197197
func (rs resultStore) printResults(
198-
gl *config.Global,
198+
gl *config.Engine,
199199
remotes config.RemoteStore,
200200
currentDSCP *ip.DSCPValue,
201201
logger *log.Logger,
@@ -209,7 +209,7 @@ func (rs resultStore) printResults(
209209

210210
// Run processes the echoes sent and received to compute and report all the metrics desired.
211211
func Run(
212-
gl *config.Global,
212+
gl *config.Engine,
213213
sentC chan tcp.Message,
214214
rcvdC chan tcp.Message,
215215
remotes config.RemoteStore,
@@ -243,7 +243,7 @@ func Run(
243243
}
244244

245245
func batchWorker(
246-
gl *config.Global,
246+
gl *config.Engine,
247247
sentC chan tcp.Message,
248248
rcvdC chan tcp.Message,
249249
remotes config.RemoteStore,
@@ -440,7 +440,7 @@ func zeroOutResults(
440440
}
441441
}
442442

443-
func printTableHeader(gl *config.Global, currentDSCP string, logger *log.Logger) {
443+
func printTableHeader(gl *config.Engine, currentDSCP string, logger *log.Logger) {
444444
color.Set(color.FgHiYellow, color.Bold)
445445
defer color.Unset()
446446

‎config/config.go

+14-9
Original file line numberDiff line numberDiff line change
@@ -39,12 +39,12 @@ import (
3939
"github.com/uber/arachne/metrics"
4040

4141
"github.com/google/gopacket/layers"
42-
"github.com/jawher/mow.cli"
42+
cli "github.com/jawher/mow.cli"
4343
"github.com/pkg/errors"
4444
"go.uber.org/zap"
4545
"go.uber.org/zap/zapcore"
46-
"gopkg.in/validator.v2"
47-
"gopkg.in/yaml.v2"
46+
validator "gopkg.in/validator.v2"
47+
yaml "gopkg.in/yaml.v2"
4848
)
4949

5050
const defaultConfigFile = "/etc/arachne/arachne.yaml"
@@ -154,14 +154,19 @@ type pollInterval struct {
154154
Failure time.Duration
155155
}
156156

157-
// Global holds the global application info.
158-
type Global struct {
159-
App *AppConfig
160-
CLI *CLIConfig
157+
// Engine helds the configuration for the polling engine.
158+
type Engine struct {
161159
RemoteConfig *RemoteConfig
162160
Remotes RemoteStore
163161
}
164162

163+
// Global holds the global application info.
164+
type Global struct {
165+
Engine
166+
App *AppConfig
167+
CLI *CLIConfig
168+
}
169+
165170
func localFileReadable(path string) error {
166171
if _, err := ioutil.ReadFile(path); err != nil {
167172
return err
@@ -280,7 +285,7 @@ func FetchRemoteList(
280285
minBatchInterval time.Duration,
281286
HTTPResponseHeaderTimeout time.Duration,
282287
orchestratorRESTConf string,
283-
kill chan struct{},
288+
kill <-chan struct{},
284289
logger *log.Logger,
285290
) error {
286291

@@ -359,7 +364,7 @@ func refreshRemoteList(
359364
minBatchInterval time.Duration,
360365
HTTPResponseHeaderTimeout time.Duration,
361366
orchestratorRESTConf string,
362-
kill chan struct{},
367+
kill <-chan struct{},
363368
logger *log.Logger,
364369
) error {
365370

0 commit comments

Comments
 (0)