This is an initial implementation of the mgmt language. It is a declarative (immutable) functional, reactive, domain specific programming language. It is intended to be a language that is: * safe * powerful * easy to reason about With these properties, we hope this language, and the mgmt engine will allow you to model the real-time systems that you'd like to automate. This also includes a number of other associated changes. Sorry for the large size of this patch.
268 lines
8.0 KiB
Go
268 lines
8.0 KiB
Go
// Mgmt
|
|
// Copyright (C) 2013-2018+ James Shubin and the project contributors
|
|
// Written by James Shubin <james@shubin.ca> and the project contributors
|
|
//
|
|
// This program is free software: you can redistribute it and/or modify
|
|
// it under the terms of the GNU General Public License as published by
|
|
// the Free Software Foundation, either version 3 of the License, or
|
|
// (at your option) any later version.
|
|
//
|
|
// This program is distributed in the hope that it will be useful,
|
|
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
// GNU General Public License for more details.
|
|
//
|
|
// You should have received a copy of the GNU General Public License
|
|
// along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
package lang // TODO: move this into a sub package of lang/$name?
|
|
|
|
import (
|
|
"fmt"
|
|
"io"
|
|
"log"
|
|
"sync"
|
|
|
|
"github.com/purpleidea/mgmt/lang/funcs"
|
|
_ "github.com/purpleidea/mgmt/lang/funcs/core" // import so the funcs register
|
|
_ "github.com/purpleidea/mgmt/lang/funcs/facts/core" // import so the facts register
|
|
"github.com/purpleidea/mgmt/lang/interfaces"
|
|
"github.com/purpleidea/mgmt/lang/unification"
|
|
"github.com/purpleidea/mgmt/pgraph"
|
|
"github.com/purpleidea/mgmt/resources"
|
|
|
|
errwrap "github.com/pkg/errors"
|
|
)
|
|
|
|
const (
|
|
// FileNameExtension is the filename extension used for languages files.
|
|
FileNameExtension = "mcl" // alternate suggestions welcome!
|
|
|
|
// make these available internally without requiring the import
|
|
operatorFuncName = funcs.OperatorFuncName
|
|
historyFuncName = funcs.HistoryFuncName
|
|
containsFuncName = funcs.ContainsFuncName
|
|
)
|
|
|
|
// Lang is the main language lexer/parser object.
|
|
type Lang struct {
|
|
Input io.Reader // os.Stdin or anything that satisfies this interface
|
|
Hostname string
|
|
World resources.World
|
|
Debug bool
|
|
|
|
ast interfaces.Stmt // store main prog AST here
|
|
funcs *funcs.Engine // function event engine
|
|
|
|
loadedChan chan struct{} // loaded signal
|
|
|
|
streamChan chan error // signals a new graph can be created or problem
|
|
//streamBurst bool // should we try and be bursty with the stream events?
|
|
|
|
closeChan chan struct{} // close signal
|
|
wg *sync.WaitGroup
|
|
}
|
|
|
|
// Init initializes the lang struct, and starts up the initial data sources.
|
|
// NOTE: The trick is that we need to get the list of funcs to watch AND start
|
|
// watching them, *before* we pull their values, that way we'll know if they
|
|
// changed from the values we wanted.
|
|
func (obj *Lang) Init() error {
|
|
obj.loadedChan = make(chan struct{})
|
|
obj.streamChan = make(chan error)
|
|
obj.closeChan = make(chan struct{})
|
|
obj.wg = &sync.WaitGroup{}
|
|
|
|
once := &sync.Once{}
|
|
loadedSignal := func() { close(obj.loadedChan) } // only run once!
|
|
|
|
// run the lexer/parser and build an AST
|
|
log.Printf("%s: Lexing/Parsing...", Name)
|
|
ast, err := LexParse(obj.Input)
|
|
if err != nil {
|
|
return errwrap.Wrapf(err, "could not generate AST")
|
|
}
|
|
if obj.Debug {
|
|
log.Printf("%s: behold, the AST: %+v", Name, ast)
|
|
}
|
|
|
|
// TODO: should we validate the structure of the AST?
|
|
// TODO: should we do this *after* interpolate, or trust it to behave?
|
|
//if err := ast.Validate(); err != nil {
|
|
// return errwrap.Wrapf(err, "could not validate AST")
|
|
//}
|
|
|
|
log.Printf("%s: Interpolating...", Name)
|
|
// interpolate strings and other expansionable nodes in AST
|
|
interpolated, err := ast.Interpolate()
|
|
if err != nil {
|
|
return errwrap.Wrapf(err, "could not interpolate AST")
|
|
}
|
|
obj.ast = interpolated
|
|
|
|
// top-level, built-in, initial global scope
|
|
scope := &interfaces.Scope{
|
|
Variables: map[string]interfaces.Expr{
|
|
"purpleidea": &ExprStr{V: "hello world!"}, // james says hi
|
|
// TODO: change to a func when we can change hostname dynamically!
|
|
"hostname": &ExprStr{V: obj.Hostname},
|
|
},
|
|
}
|
|
|
|
log.Printf("%s: Building Scope...", Name)
|
|
// propagate the scope down through the AST...
|
|
if err := obj.ast.SetScope(scope); err != nil {
|
|
return errwrap.Wrapf(err, "could not set scope")
|
|
}
|
|
|
|
// apply type unification
|
|
logf := func(format string, v ...interface{}) {
|
|
if obj.Debug { // unification only has debug messages...
|
|
log.Printf(Name+": unification: "+format, v...)
|
|
}
|
|
}
|
|
log.Printf("%s: Running Type Unification...", Name)
|
|
if err := unification.Unify(obj.ast, unification.SimpleInvariantSolverLogger(logf)); err != nil {
|
|
return errwrap.Wrapf(err, "could not unify types")
|
|
}
|
|
|
|
log.Printf("%s: Building Function Graph...", Name)
|
|
// we assume that for some given code, the list of funcs doesn't change
|
|
// iow, we don't support variable, variables or absurd things like that
|
|
graph, err := obj.ast.Graph() // build the graph of functions
|
|
if err != nil {
|
|
return errwrap.Wrapf(err, "could not generate function graph")
|
|
}
|
|
|
|
if obj.Debug {
|
|
log.Printf("%s: function graph: %+v", Name, graph)
|
|
graph.Logf("%s: ", Name) // log graph with this printf prefix...
|
|
}
|
|
|
|
if graph.NumVertices() == 0 { // no funcs to load!
|
|
// send only one signal since we won't ever send after this!
|
|
log.Printf("%s: Static graph found", Name)
|
|
obj.wg.Add(1)
|
|
go func() {
|
|
defer obj.wg.Done()
|
|
defer close(obj.streamChan) // no more events are coming!
|
|
select {
|
|
case obj.streamChan <- nil: // send one signal
|
|
// pass
|
|
case <-obj.closeChan:
|
|
return
|
|
}
|
|
close(obj.loadedChan) // signal
|
|
}()
|
|
return nil // exit early, no funcs to load!
|
|
}
|
|
|
|
obj.funcs = &funcs.Engine{
|
|
Graph: graph, // not the same as the output graph!
|
|
Hostname: obj.Hostname,
|
|
World: obj.World,
|
|
Debug: obj.Debug,
|
|
Logf: func(format string, v ...interface{}) {
|
|
log.Printf(Name+": funcs: "+format, v...)
|
|
},
|
|
Glitch: false, // FIXME: verify this functionality is perfect!
|
|
}
|
|
|
|
log.Printf("%s: Function Engine Initializing...", Name)
|
|
if err := obj.funcs.Init(); err != nil {
|
|
return errwrap.Wrapf(err, "init error with func engine")
|
|
}
|
|
|
|
log.Printf("%s: Function Engine Validating...", Name)
|
|
if err := obj.funcs.Validate(); err != nil {
|
|
return errwrap.Wrapf(err, "validate error with func engine")
|
|
}
|
|
|
|
log.Printf("%s: Function Engine Starting...", Name)
|
|
// On failure, we expect the caller to run Close() to shutdown all of
|
|
// the currently initialized (and running) funcs... This is needed if
|
|
// we successfully ran `Run` but isn't needed only for Init/Validate.
|
|
if err := obj.funcs.Run(); err != nil {
|
|
return errwrap.Wrapf(err, "run error with func engine")
|
|
}
|
|
|
|
// wait for some activity
|
|
log.Printf("%s: Stream...", Name)
|
|
stream := obj.funcs.Stream()
|
|
obj.wg.Add(1)
|
|
go func() {
|
|
log.Printf("%s: Loop...", Name)
|
|
defer obj.wg.Done()
|
|
defer close(obj.streamChan) // no more events are coming!
|
|
for {
|
|
var err error
|
|
var ok bool
|
|
select {
|
|
case err, ok = <-stream:
|
|
if !ok {
|
|
log.Printf("%s: Stream closed", Name)
|
|
return
|
|
}
|
|
if err == nil {
|
|
// only do this once, on the first event
|
|
once.Do(loadedSignal) // signal
|
|
}
|
|
|
|
case <-obj.closeChan:
|
|
return
|
|
}
|
|
|
|
select {
|
|
case obj.streamChan <- err: // send
|
|
if err != nil {
|
|
log.Printf("%s: Stream error: %+v", Name, err)
|
|
return
|
|
}
|
|
|
|
case <-obj.closeChan:
|
|
return
|
|
}
|
|
}
|
|
}()
|
|
return nil
|
|
}
|
|
|
|
// Stream returns a channel of graph change requests or errors. These are
|
|
// usually sent when a func output changes.
|
|
func (obj *Lang) Stream() chan error {
|
|
return obj.streamChan
|
|
}
|
|
|
|
// Interpret runs the interpreter and returns a graph and corresponding error.
|
|
func (obj *Lang) Interpret() (*pgraph.Graph, error) {
|
|
select {
|
|
case <-obj.loadedChan: // funcs are now loaded!
|
|
// pass
|
|
default:
|
|
// if this is hit, someone probably called this too early!
|
|
// it should only be called in response to a stream event!
|
|
return nil, fmt.Errorf("funcs aren't loaded yet")
|
|
}
|
|
|
|
log.Printf("%s: Running interpret...", Name)
|
|
// this call returns the graph
|
|
graph, err := interpret(obj.ast)
|
|
if err != nil {
|
|
return nil, errwrap.Wrapf(err, "could not interpret")
|
|
}
|
|
|
|
return graph, nil // return a graph
|
|
}
|
|
|
|
// Close shuts down the lang struct and causes all the funcs to shutdown. It
|
|
// must be called when finished after any successful Init ran.
|
|
func (obj *Lang) Close() error {
|
|
var err error
|
|
if obj.funcs != nil {
|
|
err = obj.funcs.Close()
|
|
}
|
|
close(obj.closeChan)
|
|
obj.wg.Wait()
|
|
return err
|
|
}
|