forked from gravwell/migrate
-
Notifications
You must be signed in to change notification settings - Fork 0
/
config.go
170 lines (155 loc) · 4.16 KB
/
config.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
/*************************************************************************
* Copyright 2022 Gravwell, Inc. All rights reserved.
* Contact: <[email protected]>
*
* This software may be modified and distributed under the terms of the
* BSD 2-clause license. See the LICENSE file for details.
**************************************************************************/
package main
import (
"errors"
"fmt"
"sort"
"github.com/google/uuid"
"github.com/gravwell/gravwell/v3/ingest"
"github.com/gravwell/gravwell/v3/ingest/config"
"github.com/gravwell/gravwell/v3/ingest/entry"
"github.com/gravwell/gravwell/v3/ingest/processors"
)
var (
ErrInvalidStateStoreLocation = errors.New("Empty state storage location")
ErrTimestampDelimiterMissingOverride = errors.New("Timestamp delimiting requires a defined timestamp override")
)
type bindType int
type readerType int
type cfgReadType struct {
Global global
Files map[string]*files
Splunk map[string]*splunk
Preprocessor processors.ProcessorConfig
TimeFormat config.CustomTimeFormat
}
type global struct {
config.IngestConfig
State_Store_Location string
}
type cfgType struct {
global
Files map[string]*files
Splunk map[string]*splunk
Preprocessor processors.ProcessorConfig
TimeFormat config.CustomTimeFormat
}
func GetConfig(path, overlayPath string) (*cfgType, error) {
var cr cfgReadType
if err := config.LoadConfigFile(&cr, path); err != nil {
return nil, err
} else if err = config.LoadConfigOverlays(&cr, overlayPath); err != nil {
return nil, err
}
c := &cfgType{
global: cr.Global,
Files: cr.Files,
Splunk: cr.Splunk,
Preprocessor: cr.Preprocessor,
TimeFormat: cr.TimeFormat,
}
if err := verifyConfig(c); err != nil {
return nil, err
}
// Verify and set UUID
if _, ok := c.IngesterUUID(); !ok {
id := uuid.New()
if err := c.SetIngesterUUID(id, path); err != nil {
return nil, err
}
if id2, ok := c.IngesterUUID(); !ok || id != id2 {
return nil, errors.New("Failed to set a new ingester UUID")
}
}
return c, nil
}
func verifyConfig(c *cfgType) error {
//verify the global parameters
if err := c.Verify(); err != nil {
return err
}
if len(c.Files) == 0 && len(c.Splunk) == 0 {
return errors.New("No Files or Splunk stanzas specified")
}
if err := c.Preprocessor.Validate(); err != nil {
return err
} else if err = c.TimeFormat.Validate(); err != nil {
return err
}
for k, v := range c.Files {
if err := v.Validate(c.Preprocessor); err != nil {
return fmt.Errorf("Files config %s failed %w", k, err)
}
}
for k, v := range c.Splunk {
if err := v.Validate(c.Preprocessor); err != nil {
return fmt.Errorf("Splunk config %s failed %w", k, err)
}
}
return nil
}
// Tags returns a list of tags specified in the config.
// It will always include the gravwell tag.
func (c *cfgType) Tags() ([]string, error) {
tags := []string{entry.GravwellTagName}
tagMp := make(map[string]bool, 1)
tagMp[entry.GravwellTagName] = true
for _, v := range c.Files {
if len(v.Tag_Name) == 0 {
continue
}
if _, ok := tagMp[v.Tag_Name]; !ok {
tags = append(tags, v.Tag_Name)
tagMp[v.Tag_Name] = true
}
}
for _, v := range c.Splunk {
if tgs, err := v.Tags(); err != nil {
return tags, err
} else {
for _, tag := range tgs {
tags = append(tags, tag)
tagMp[tag] = true
}
}
}
sort.Strings(tags)
return tags, nil
}
func (c *cfgType) getSplunkConn(splunkName string) (sc splunkConn, err error) {
for k, vv := range c.Splunk {
if k == splunkName {
// Connect to Splunk server
sc = newSplunkConn(vv.Server, vv.Token)
return
}
}
err = errors.New("Not found")
return
}
func (c *cfgType) getSplunkPreprocessors(splunkName string, igst *ingest.IngestMuxer) (pproc *processors.ProcessorSet, err error) {
for k, vv := range c.Splunk {
if k == splunkName {
// get the ingester up and rolling
pproc, err = cfg.Preprocessor.ProcessorSet(igst, vv.Preprocessor)
return
}
}
err = errors.New("Not found")
return
}
func (g *global) Verify() (err error) {
if err = g.IngestConfig.Verify(); err != nil {
return
}
return
}
func (g *global) StatePath() string {
return g.State_Store_Location
}