Run configure and opens the event source to start consuming events. Depending on whether the filament is provided, this method will either spin up a filament or set up the aggregator to start forwarding events to the rule engine and output sinks.
(args []string)
| 172 | // spin up a filament or set up the aggregator to start forwarding events |
| 173 | // to the rule engine and output sinks. |
| 174 | func (f *App) Run(args []string) error { |
| 175 | if f.evs == nil { |
| 176 | panic("event source is nil") |
| 177 | } |
| 178 | cfg := f.config |
| 179 | |
| 180 | if !f.isSingleInstance() { |
| 181 | return ErrAlreadyRunning |
| 182 | } |
| 183 | |
| 184 | log.Infof("bootstrapping with pid %d. Version: %s", os.Getpid(), version.Get()) |
| 185 | log.Infof("configuration options: %s", cfg.Print()) |
| 186 | |
| 187 | // build the filter from the CLI argument. If we got |
| 188 | // a valid expression the filter is attached to the |
| 189 | // event consumer |
| 190 | fltr, err := filter.NewFromCLI(args, cfg) |
| 191 | if err != nil { |
| 192 | return err |
| 193 | } |
| 194 | if fltr != nil { |
| 195 | f.evs.SetFilter(fltr) |
| 196 | } |
| 197 | // user can either instruct to bootstrap a filament or |
| 198 | // start a regular run. We'll set up the corresponding |
| 199 | // components accordingly to what we got from the CLI options. |
| 200 | // If a filament was given, we'll assign it the previous filter |
| 201 | // if it wasn't provided in the filament init function. |
| 202 | // Finally, we open the event source and run the filament i.e. |
| 203 | // Python main thread in a new goroutine. |
| 204 | // In case of a regular run, we additionally set up the aggregator. |
| 205 | // The aggregator will grab the events from the queue, assemble them |
| 206 | // into batches and hand over to output sinks. |
| 207 | if cfg.IsFilamentSet() { |
| 208 | f.filament, err = filament.New(cfg.Filament.Name, f.psnap, f.hsnap, cfg) |
| 209 | if err != nil { |
| 210 | return err |
| 211 | } |
| 212 | if f.filament.Filter() != nil { |
| 213 | f.evs.SetFilter(f.filament.Filter()) |
| 214 | } |
| 215 | err = f.evs.Open(cfg) |
| 216 | if err != nil { |
| 217 | return multierror.Wrap(err, f.evs.Close()) |
| 218 | } |
| 219 | // load alert senders so emitting alerts is possible from filaments |
| 220 | err = alertsender.LoadAll(cfg.Alertsenders) |
| 221 | if err != nil { |
| 222 | log.Warnf("couldn't load alertsenders: %v", err) |
| 223 | } |
| 224 | go func() { |
| 225 | err = f.filament.Run(f.evs.Events(), f.evs.Errors()) |
| 226 | if err != nil { |
| 227 | log.Errorf("filament failed: %v", err) |
| 228 | f.stop() |
| 229 | } |
| 230 | }() |
| 231 | } else { |
no test coverage detected