// Mgmt // Copyright (C) 2013-2018+ James Shubin and the project contributors // Written by James Shubin 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 . 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 }