Skip to content

Commit

Permalink
new mysql storage backend for workflow engine (#33)
Browse files Browse the repository at this point in the history
  • Loading branch information
jessepeterson authored Oct 20, 2023
1 parent fe46327 commit 3f97c32
Show file tree
Hide file tree
Showing 28 changed files with 2,362 additions and 60 deletions.
42 changes: 42 additions & 0 deletions .github/workflows/on-push-pr.yml
Original file line number Diff line number Diff line change
Expand Up @@ -83,3 +83,45 @@ jobs:
with:
name: release-zips
path: "*.zip"
mysql-test:
runs-on: 'ubuntu-latest'
needs: format-build-test
services:
mysql:
image: mysql:8.0
env:
MYSQL_RANDOM_ROOT_PASSWORD: yes
MYSQL_DATABASE: nanocmd
MYSQL_USER: nanocmd
MYSQL_PASSWORD: nanocmd
ports:
- 3800:3306
options: --health-cmd="mysqladmin ping" --health-interval=5s --health-timeout=2s --health-retries=3
defaults:
run:
shell: bash
env:
MYSQL_PWD: nanocmd
PORT: 3800
steps:
- uses: actions/checkout@8ade135a41bc03ea155e62e844d188df1ea18608 # v4.1.0

- uses: actions/setup-go@6edd4406fa81c3da01a34fa6f6343087c207a568 # v3.5.0
with:
go-version: '1.19.x'

- name: verify mysql
run: |
while ! mysqladmin ping --host=localhost --port=$PORT --protocol=TCP --silent; do
sleep 1
done
- name: mysql schema
run: |
mysql --version
mysql --user=nanocmd --host=localhost --port=$PORT --protocol=TCP nanocmd < ./engine/storage/mysql/schema.sql
- name: setup test dsn
run: echo "NANOCMD_MYSQL_STORAGE_TEST_DSN=nanocmd:nanocmd@tcp(localhost:$PORT)/nanocmd" >> $GITHUB_ENV

- run: go test -v ./engine/storage/mysql
21 changes: 21 additions & 0 deletions cmd/nanocmd/storage.go
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ import (
storageeng "github.com/micromdm/nanocmd/engine/storage"
storageengdiskv "github.com/micromdm/nanocmd/engine/storage/diskv"
storageenginmem "github.com/micromdm/nanocmd/engine/storage/inmem"
storageengmysql "github.com/micromdm/nanocmd/engine/storage/mysql"
storagecmdplan "github.com/micromdm/nanocmd/subsystem/cmdplan/storage"
storagecmdplandiskv "github.com/micromdm/nanocmd/subsystem/cmdplan/storage/diskv"
storagecmdplaninmem "github.com/micromdm/nanocmd/subsystem/cmdplan/storage/inmem"
Expand All @@ -19,6 +20,8 @@ import (
storageprof "github.com/micromdm/nanocmd/subsystem/profile/storage"
storageprofdiskv "github.com/micromdm/nanocmd/subsystem/profile/storage/diskv"
storageprofinmem "github.com/micromdm/nanocmd/subsystem/profile/storage/inmem"

_ "github.com/go-sql-driver/mysql"
)

type storageConfig struct {
Expand Down Expand Up @@ -65,6 +68,24 @@ func parseStorage(name, dsn string) (*storageConfig, error) {
event: eng,
filevault: fv,
}, nil
case "mysql":
inv := storageinvinmem.New()
fv, err := storagefvinmem.New(storagefvinvprk.NewInvPRK(inv))
if err != nil {
return nil, fmt.Errorf("creating filevault inmem storage: %w", err)
}
eng, err := storageengmysql.New(storageengmysql.WithDSN(dsn))
if err != nil {
return nil, err
}
return &storageConfig{
engine: eng,
inventory: inv,
profile: storageprofinmem.New(),
cmdplan: storagecmdplaninmem.New(),
event: eng,
filevault: fv,
}, nil
}
return nil, fmt.Errorf("unknown storage: %s", name)
}
11 changes: 11 additions & 0 deletions docs/operations-guide.md
Original file line number Diff line number Diff line change
Expand Up @@ -88,6 +88,17 @@ Configures the `inmem` storage backend. Data is stored entirely in-memory and is

*Example:* `-storage inmem`

##### mysql storage backend

* `-storage mysql`

Configures the MySQL storage backend. The `-storage-dsn` flag should be in the [format the SQL driver expects](https://github.com/go-sql-driver/mysql#dsn-data-source-name).
Be sure to create the storage tables with the [schema.sql](../storage/mysql/schema.sql) file. MySQL 8.0.19 or later is required.

**WARNING:** The MySQL backend currently only implements storage for the workflow *engine*. When running NanoCMD the *subsystem* storage is completely in-memory as if you supplied `-storage inmem`. The practical effect is that subsystem storage is volatile and no data will be persisted for them.

*Example:* `-storage mysql -dsn nanocmd:nanocmd/mycmddb`

#### -version

* print version
Expand Down
2 changes: 1 addition & 1 deletion engine/convert.go
Original file line number Diff line number Diff line change
Expand Up @@ -129,7 +129,7 @@ func storageStepCommandFromRawResponse(reqType string, rawResp []byte) (*storage
ResultReport: rawResp,
Completed: genResp.Status != "" && genResp.Status != "NotNow",
}
return sc, response, nil
return sc, response, sc.Validate()
}

// workflowCommandResponseFromRawResponse converts a raw XML plist of a command response to a workflow response.
Expand Down
21 changes: 21 additions & 0 deletions engine/storage/kv/kv.go
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ package kv

import (
"context"
"errors"
"fmt"
"sync"
"time"
Expand Down Expand Up @@ -33,6 +34,9 @@ func New(stepStore kv.TraversingBucket, idCmdStore kv.TraversingBucket, eventSto

// RetrieveCommandRequestType implements the storage interface method.
func (s *KV) RetrieveCommandRequestType(ctx context.Context, id string, cmdUUID string) (string, bool, error) {
if id == "" || cmdUUID == "" {
return "", false, errors.New("empty id or command uuid")
}
s.mu.RLock()
defer s.mu.RUnlock()
// first check if we have a valid command
Expand Down Expand Up @@ -157,6 +161,23 @@ func (s *KV) StoreStep(ctx context.Context, step *storage.StepEnqueuingWithConfi
// fabricate a unique ID to track this unique step
stepID := s.ider.ID()

if step != nil {
idCmdUUIDs := make(map[string]struct{})
for _, sc := range step.Commands {
for _, id := range step.IDs {
if _, ok := idCmdUUIDs[id+sc.CommandUUID]; ok {
return fmt.Errorf("duplicate command (id=%s, uuid=%s)", id, sc.CommandUUID)
}
idCmdUUIDs[id+sc.CommandUUID] = struct{}{}
if ok, err := kvIDCmdExists(ctx, s.idCmdStore, id, sc.CommandUUID); err != nil {
return fmt.Errorf("checking duplicate commands: %w", err)
} else if ok {
return fmt.Errorf("duplicate command (id=%s, uuid=%s)", id, sc.CommandUUID)
}
}
}
}

err := kvSetStep(ctx, s.stepStore, stepID, step)
if err != nil {
return fmt.Errorf("setting step record: %w", err)
Expand Down
74 changes: 74 additions & 0 deletions engine/storage/mysql/event.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,74 @@
package mysql

import (
"context"
"fmt"

"github.com/micromdm/nanocmd/engine/storage"
"github.com/micromdm/nanocmd/workflow"
)

// RetrieveEventSubscriptions retrieves event subscriptions by names.
// See the storage interface type for further docs.
func (s *MySQLStorage) RetrieveEventSubscriptions(ctx context.Context, names []string) (map[string]*storage.EventSubscription, error) {
events, err := s.q.GetEventsByNames(ctx, names)
if err != nil {
return nil, fmt.Errorf("get events by name: %w", err)
}
retEvents := make(map[string]*storage.EventSubscription)
for _, event := range events {
retEvents[event.EventName] = &storage.EventSubscription{
Event: event.EventType,
Workflow: event.WorkflowName,
Context: event.Context.String,
}
}
return retEvents, nil
}

// RetrieveEventSubscriptionsByEvent retrieves event subscriptions by event flag.
// See the storage interface type for further docs.
func (s *MySQLStorage) RetrieveEventSubscriptionsByEvent(ctx context.Context, f workflow.EventFlag) ([]*storage.EventSubscription, error) {
events, err := s.q.GetEventsByType(ctx, f.String())
if err != nil {
return nil, fmt.Errorf("get events by type: %w", err)
}
var retEvents []*storage.EventSubscription
for _, event := range events {
retEvents = append(retEvents, &storage.EventSubscription{
Event: event.EventType,
Workflow: event.WorkflowName,
Context: event.Context.String,
})
}
return retEvents, nil
}

// StoreEventSubscription stores an event subscription.
// See the storage interface type for further docs.
func (s *MySQLStorage) StoreEventSubscription(ctx context.Context, name string, es *storage.EventSubscription) error {
_, err := s.db.ExecContext(
ctx,
`
INSERT INTO wf_events
(event_name, event_type, workflow_name, context)
VALUES
(?, ?, ?, ?) AS new
ON DUPLICATE KEY
UPDATE
workflow_name = new.workflow_name,
event_type = new.event_type,
context = new.context;`,
name,
es.Event,
es.Workflow,
sqlNullString(es.Context),
)
return err
}

// DeleteEventSubscription removes an event subscription.
// See the storage interface type for further docs.
func (s *MySQLStorage) DeleteEventSubscription(ctx context.Context, name string) error {
return s.q.RemoveEvent(ctx, name)
}
3 changes: 3 additions & 0 deletions engine/storage/mysql/generate.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
package mysql

//go:generate sqlc generate
108 changes: 108 additions & 0 deletions engine/storage/mysql/mysql.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,108 @@
package mysql

import (
"context"
"database/sql"
"fmt"
"math/rand"
"sync"
"time"

"github.com/micromdm/nanocmd/engine/storage/mysql/sqlc"
)

// MySQLStorage implements a storage.AllStorage using MySQL.
type MySQLStorage struct {
db *sql.DB
q *sqlc.Queries

randMu sync.Mutex
rand *rand.Rand
}

type config struct {
driver string
dsn string
db *sql.DB
}

// Option allows configuring a MySQLStorage.
type Option func(*config)

// WithDSN sets the storage MySQL data source name.
func WithDSN(dsn string) Option {
return func(c *config) {
c.dsn = dsn
}
}

// WithDriver sets a custom MySQL driver for the storage.
// Default driver is "mysql" but is ignored if WithDB is used.
func WithDriver(driver string) Option {
return func(c *config) {
c.driver = driver
}
}

// WithDB sets a custom MySQL *sql.DB to the storage.
// If set, driver passed via WithDriver is ignored.
func WithDB(db *sql.DB) Option {
return func(c *config) {
c.db = db
}
}

// New creates and returns a new MySQL.
func New(opts ...Option) (*MySQLStorage, error) {
cfg := &config{driver: "mysql"}
for _, opt := range opts {
opt(cfg)
}
var err error
if cfg.db == nil {
cfg.db, err = sql.Open(cfg.driver, cfg.dsn)
if err != nil {
return nil, err
}
}
if err = cfg.db.Ping(); err != nil {
return nil, err
}
return &MySQLStorage{
db: cfg.db,
q: sqlc.New(cfg.db),
rand: rand.New(rand.NewSource(time.Now().UnixNano())),
}, nil
}

// sqlNullString sets Valid to true of the return value of s is not empty.
func sqlNullString(s string) sql.NullString {
return sql.NullString{String: s, Valid: s != ""}
}

// sqlNullTime sets Valid to true of the return value of t is not zero.
func sqlNullTime(t time.Time) sql.NullTime {
return sql.NullTime{Valid: !t.IsZero(), Time: t}
}

// txcb executes SQL within transactions when wrapped in tx().
type txcb func(ctx context.Context, tx *sql.Tx, qtx *sqlc.Queries) error

// tx wraps g in transactions using db.
// If g returns an err the transaction will be rolled back; otherwise committed.
func tx(ctx context.Context, db *sql.DB, q *sqlc.Queries, g txcb) error {
tx, err := db.BeginTx(ctx, nil)
if err != nil {
return fmt.Errorf("tx begin: %w", err)
}
if err = g(ctx, tx, q.WithTx(tx)); err != nil {
if rbErr := tx.Rollback(); rbErr != nil {
return fmt.Errorf("tx rollback: %w; while trying to handle error: %v", rbErr, err)
}
return fmt.Errorf("tx rolled back: %w", err)
}
if err = tx.Commit(); err != nil {
return fmt.Errorf("tx commit: %w", err)
}
return nil
}
25 changes: 25 additions & 0 deletions engine/storage/mysql/mysql_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,25 @@
package mysql

import (
"os"
"testing"

"github.com/micromdm/nanocmd/engine/storage"
"github.com/micromdm/nanocmd/engine/storage/test"

_ "github.com/go-sql-driver/mysql"
)

func TestMySQLStorage(t *testing.T) {
testDSN := os.Getenv("NANOCMD_MYSQL_STORAGE_TEST_DSN")
if testDSN == "" {
t.Skip("NANOCMD_MYSQL_STORAGE_TEST_DSN not set")
}

s, err := New(WithDSN(testDSN))
if err != nil {
t.Fatal(err)
}

test.TestEngineStorage(t, func() storage.AllStorage { return s })
}
Loading

0 comments on commit 3f97c32

Please sign in to comment.