feat: add backing store for disk buffering of events #273
Conversation
notque
requested review from
Nuckal777,
SuperSandro2000,
VoigtS,
majewsky and
wagnerd3
as code owners
notque
force-pushed
the
audit_backing_store
branch
from
d1887ee to
6e065d1
Compare
notque
force-pushed
the
audit_backing_store
branch
from
6e065d1 to
8d68fb5
Compare
There was a problem hiding this comment.
Dealing with filesystem volumes is going to be a major complication for all of my services, where I do not have volumes at all. I have already considered writing audit events into the DB until they can be submitted, so if we introduce a caching capability at this level, I would like for it to be able to support a DB as backing store, too.
I can see that you have type BackingStore as an interface, so presumably we can provide an SQL-based implementation instead. We don't need to have this implementation as part of this PR, in order to keep it at a manageable size. But what this PR should do is build the public interface in such a way that it allows specifying backing stores other than FileBackingStore, potentially with different types of configuration parameters.
Since we need to support passing configuration via env variables, I suggest something similar to how we pass configuration for Keppel drivers. This is a full example, and this is how we parse these, but basically we could have something like this:
export ${PREFIX}_BACKING_STORE='{"type":"fs","params":{"path":"/var/cache/audit","max_total_size":1073741824}}'The difference between opts.BackingStoreFactories vs. opts.BackingStore is similar to opts.EnvPrefix vs. opts.ConnectionURL: One allows using the default logic of collecting everything from env vars, one allows the application precise control over where to collect config from.
To allow both the application as well as this library to provide BackingStore implementations, I suggest modeling AuditorOpts like this:
type AuditorOpts struct {
// Optional. If given, this BackingStore instance will be used directly.
// If EnvPrefix is given, this will be initialized by reading a JSON payload in the form `{"type":"<type>","params":{...}}`
// from the environment variable "${PREFIX}_BACKING_STORE".
BackingStore BackingStore
// Optional. If given, and the environment contains JSON configuration as described above,
// a BackingStore constructor will be selected from this set based on the configured type.
BackingStoreFactories map[string]BackingStoreFactory
}
type BackingStoreFactory func(params json.RawMessage, opts AuditorOpts) (BackingStore, error)
func NewFileBackingStore(params json.RawMessage, opts AuditorOpts) (BackingStore, error) {
var bsOpts struct {
Directory string `json:"path"`
MaxFileSize int64 `json:"max_file_size"`
MaxTotalSize int64 `json:"max_total_size"`
}
err := json.Unmarshal([]byte(params), &bsOpts)
if err != nil {
return nil, fmt.Errorf("while unmarshaling params for FileBackingStore: %w", err)
}
registry := opts.Registry
//... continue with existing implementation...
}Then this could be used as:
auditor := must.Return(audittools.NewAuditor(ctx, audittools.AuditorOpts{
EnvPrefix: "LIMES_AUDIT_RABBITMQ",
BackingStoreFactories: map[string]audittools.BackingStoreFactory{
"fs": audittools.NewFileBackingStore,
"db": func(params json.RawMessage, opts audittools.AuditorOpts) {
return newDBBackingStore(dbConnection, params, opts.Registry)
},
},
})What do you think?
audittools/trail.go
Outdated
| drainTicker := time.NewTicker(1 * time.Minute) | ||
| defer drainTicker.Stop() | ||
|
|
||
| var pendingEvents []cadf.Event |
There was a problem hiding this comment.
I have not read this in detail, but my immediate thought is that pendingEvents is sort-of redundant with the backing store. Both are used as a cache for events that are currently undeliverable. Would it make sense to subsume pendingEvents into the new mechanism, i.e. removing it here and instead adding an InMemoryBackingStore, that acts as a fallback if no other BackingStore is defined?
There was a problem hiding this comment.
yeah, you are correct. I've made adjustments to correctly handle this.
You bring up very reasonable points, and ones I thought you'd bring up. I'm not particularly happy with adding volumes to every service, and agree that supporting additional options like a database is reasonable. I think you understand the situation, I don't particularly think this is a great achievement, but it's a requirement I cannot avoid, and thus I am trying to bring about a solution that involves as little pain as possible. I did accidently submit this as as pr ready to review, and then set it to work in progress. Apologies you're receiving notifications from it. I am not at a stage yet where it is ready to review. I will end up pushing up a lot of changes again tonight, and then also trying to incorporate your suggestions with the state I have it in currently. I do very much appreciate your stating clearly what you want here, as I'm happy to do whatever I can to not make this terrible for you. |
notque
force-pushed
the
audit_backing_store
branch
from
8d68fb5 to
bdd7e07
Compare
|
@majewsky i've made adjustments based on your comments. Is this remotely reasonable for you. I clearly have more work to do going through feedback from copilot, but in broad strokes is this okay? |
| if s.TableName == "" { | ||
| s.TableName = "audit_events" | ||
| } | ||
| if s.BatchSize == 0 { | ||
| s.BatchSize = 100 | ||
| } | ||
| if s.MaxEvents == 0 { | ||
| s.MaxEvents = 10000 | ||
| } |
There was a problem hiding this comment.
For optional values with defaults, consider declaring the field as Option[] and then using it as s.TableName.UnwrapOr("audit_events") etc.
audittools/backing_store_sql_test.go
Outdated
| func getTestDSN(t *testing.T) string { | ||
| t.Helper() | ||
|
|
||
| // Check for test database environment variable | ||
| dsn := os.Getenv("AUDITTOOLS_TEST_DB_DSN") | ||
| if dsn == "" { | ||
| t.Skip("AUDITTOOLS_TEST_DB_DSN not set, skipping SQL backing store tests. " + | ||
| "Set to a PostgreSQL connection string to run these tests, e.g.: " + | ||
| "postgres://user:password@localhost:5432/testdb?sslmode=disable") | ||
| } | ||
|
|
||
| return dsn | ||
| } |
There was a problem hiding this comment.
Should use WithTestDB and ConnectForTest from go-bits/easypg, to allow those tests to run in the go-makefile-maker CI pipeline.
notque
force-pushed
the
audit_backing_store
branch
from
bdd7e07 to
10667c9
Compare
notque
force-pushed
the
audit_backing_store
branch
from
10667c9 to
231f191
Compare
notque
force-pushed
the
audit_backing_store
branch
from
231f191 to
f6d00c3
Compare
Merging this branch will increase overall coverage
Coverage by fileChanged files (no unit tests)
Please note that the "Total", "Covered", and "Missed" counts above refer to code statements instead of lines of code. The value in brackets refers to the test coverage of that file in the old version of the code. Changed unit test files
|
an attempt to add disk buffering of events to make the rabbitmq connection non-blocking with disk backed storage to hold if it is down. there was a single downtime this year where the network connection from scaleout had an issue, and a service was down due to it.
we cannot lose events, so we currently just stop the service. but we also have availability requirements.
thus the idea is we add pvcs to services, and hold events there with somewhat largely configured settings to withstand any connectivity issues. i do have the defaults set somewhat small at the moment.
you may dislike this, and have a significantly better plan. i'd love to hear it. i just gave it a shot.