Skip to content

Commit fa3bbbd

Browse files
author
Shawn Hurley
authored
Adding runner package. (#472)
1 parent 9398af6 commit fa3bbbd

File tree

4 files changed

+686
-0
lines changed

4 files changed

+686
-0
lines changed

‎pkg/ansible/runner/eventapi/eventapi.go

Lines changed: 184 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,184 @@
1+
// Copyright 2018 The Operator-SDK Authors
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package eventapi
16+
17+
import (
18+
"encoding/json"
19+
"fmt"
20+
"io"
21+
"io/ioutil"
22+
"net"
23+
"net/http"
24+
"strings"
25+
"sync"
26+
"time"
27+
28+
"github.com/sirupsen/logrus"
29+
)
30+
31+
// EventReceiver serves the event API
32+
type EventReceiver struct {
33+
// Events is the channel used by the event API handler to send JobEvents
34+
// back to the runner, or whatever code is using this receiver.
35+
Events chan JobEvent
36+
37+
// SocketPath is the path on the filesystem to a unix streaming socket
38+
SocketPath string
39+
40+
// URLPath is the path portion of the url at which events should be
41+
// received. For example, "/events/"
42+
URLPath string
43+
44+
// server is the http.Server instance that serves the event API. It must be
45+
// closed.
46+
server io.Closer
47+
48+
// stopped indicates if this receiver has permanently stopped receiving
49+
// events. When true, requests to POST an event will receive a "410 Gone"
50+
// response, and the body will be ignored.
51+
stopped bool
52+
53+
// mutex controls access to the "stopped" bool above, ensuring that writes
54+
// are goroutine-safe.
55+
mutex sync.RWMutex
56+
57+
// ident is the unique identifier for a particular run of ansible-runner
58+
ident string
59+
60+
// logger holds a logger that has some fields already set
61+
logger logrus.FieldLogger
62+
}
63+
64+
func New(ident string, errChan chan<- error) (*EventReceiver, error) {
65+
sockPath := fmt.Sprintf("/tmp/ansibleoperator-%s", ident)
66+
listener, err := net.Listen("unix", sockPath)
67+
if err != nil {
68+
return nil, err
69+
}
70+
71+
rec := EventReceiver{
72+
Events: make(chan JobEvent, 1000),
73+
SocketPath: sockPath,
74+
URLPath: "/events/",
75+
ident: ident,
76+
logger: logrus.WithFields(logrus.Fields{
77+
"component": "eventapi",
78+
"job": ident,
79+
}),
80+
}
81+
82+
mux := http.NewServeMux()
83+
mux.HandleFunc(rec.URLPath, rec.handleEvents)
84+
srv := http.Server{Handler: mux}
85+
rec.server = &srv
86+
87+
go func() {
88+
errChan <- srv.Serve(listener)
89+
}()
90+
return &rec, nil
91+
}
92+
93+
// Close ensures that appropriate resources are cleaned up, such as any unix
94+
// streaming socket that may be in use. Close must be called.
95+
func (e *EventReceiver) Close() {
96+
e.mutex.Lock()
97+
e.stopped = true
98+
e.mutex.Unlock()
99+
e.logger.Debug("event API stopped")
100+
e.server.Close()
101+
close(e.Events)
102+
}
103+
104+
func (e *EventReceiver) handleEvents(w http.ResponseWriter, r *http.Request) {
105+
if r.URL.Path != e.URLPath {
106+
http.NotFound(w, r)
107+
e.logger.WithFields(logrus.Fields{
108+
"code": "404",
109+
}).Infof("path not found: %s\n", r.URL.Path)
110+
return
111+
}
112+
113+
if r.Method != http.MethodPost {
114+
e.logger.WithFields(logrus.Fields{
115+
"code": "405",
116+
}).Infof("method %s not allowed", r.Method)
117+
w.WriteHeader(http.StatusMethodNotAllowed)
118+
return
119+
}
120+
121+
ct := r.Header.Get("content-type")
122+
if strings.Split(ct, ";")[0] != "application/json" {
123+
e.logger.WithFields(logrus.Fields{
124+
"code": "415",
125+
}).Info("wrong content type: %s", ct)
126+
w.WriteHeader(http.StatusUnsupportedMediaType)
127+
w.Write([]byte("The content-type must be \"application/json\""))
128+
return
129+
}
130+
131+
body, err := ioutil.ReadAll(r.Body)
132+
if err != nil {
133+
e.logger.WithFields(logrus.Fields{
134+
"code": "500",
135+
}).Errorf("%s", err.Error())
136+
w.WriteHeader(http.StatusInternalServerError)
137+
return
138+
}
139+
140+
event := JobEvent{}
141+
err = json.Unmarshal(body, &event)
142+
if err != nil {
143+
e.logger.WithFields(logrus.Fields{
144+
"code": "400",
145+
}).Infof("could not deserialize body: %s", err.Error())
146+
w.WriteHeader(http.StatusBadRequest)
147+
w.Write([]byte("Could not deserialize body as JSON"))
148+
return
149+
}
150+
151+
// Guarantee that the Events channel will not be written to if stopped ==
152+
// true, because in that case the channel has been closed.
153+
e.mutex.RLock()
154+
defer e.mutex.RUnlock()
155+
if e.stopped {
156+
e.mutex.RUnlock()
157+
w.WriteHeader(http.StatusGone)
158+
e.logger.WithFields(logrus.Fields{
159+
"code": "410",
160+
}).Info("stopped and not accepting additional events for this job")
161+
return
162+
}
163+
// ansible-runner sends "status events" and "ansible events". The "status
164+
// events" signify a change in the state of ansible-runner itself, which
165+
// we're not currently interested in.
166+
// https://ansible-runner.readthedocs.io/en/latest/external_interface.html#event-structure
167+
if event.UUID == "" {
168+
e.logger.Info("dropping event that is not a JobEvent")
169+
} else {
170+
// timeout if the channel blocks for too long
171+
timeout := time.NewTimer(10 * time.Second)
172+
select {
173+
case e.Events <- event:
174+
case <-timeout.C:
175+
e.logger.WithFields(logrus.Fields{
176+
"code": "500",
177+
}).Warn("timed out writing event to channel")
178+
w.WriteHeader(http.StatusInternalServerError)
179+
return
180+
}
181+
_ = timeout.Stop()
182+
}
183+
w.WriteHeader(http.StatusNoContent)
184+
}

‎pkg/ansible/runner/eventapi/types.go

Lines changed: 76 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,76 @@
1+
// Copyright 2018 The Operator-SDK Authors
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package eventapi
16+
17+
import (
18+
"fmt"
19+
"strings"
20+
"time"
21+
)
22+
23+
// EventTime - time to unmarshal nano time.
24+
type EventTime struct {
25+
time.Time
26+
}
27+
28+
// UnmarshalJSON - override unmarshal json.
29+
func (e *EventTime) UnmarshalJSON(b []byte) (err error) {
30+
e.Time, err = time.Parse("2006-01-02T15:04:05.999999999", strings.Trim(string(b[:]), "\"\\"))
31+
if err != nil {
32+
return err
33+
}
34+
return nil
35+
}
36+
37+
// MarshalJSON - override the marshal json.
38+
func (e EventTime) MarshalJSON() ([]byte, error) {
39+
return []byte(fmt.Sprintf("\"%s\"", e.Time.Format("2006-01-02T15:04:05.99999999"))), nil
40+
}
41+
42+
// JobEvent - event of an ansible run.
43+
type JobEvent struct {
44+
UUID string `json:"uuid"`
45+
Counter int `json:"counter"`
46+
StdOut string `json:"stdout"`
47+
StartLine int `json:"start_line"`
48+
EndLine int `json:"EndLine"`
49+
Event string `json:"event"`
50+
EventData map[string]interface{} `json:"event_data"`
51+
PID int `json:"pid"`
52+
Created EventTime `json:"created"`
53+
}
54+
55+
// StatusJobEvent - event of an ansible run.
56+
type StatusJobEvent struct {
57+
UUID string `json:"uuid"`
58+
Counter int `json:"counter"`
59+
StdOut string `json:"stdout"`
60+
StartLine int `json:"start_line"`
61+
EndLine int `json:"EndLine"`
62+
Event string `json:"event"`
63+
EventData StatsEventData `json:"event_data"`
64+
PID int `json:"pid"`
65+
Created EventTime `json:"created"`
66+
}
67+
68+
// StatsEventData - data for a the status event.
69+
type StatsEventData struct {
70+
Playbook string `json:"playbook"`
71+
PlaybookUUID string `json:"playbook_uuid"`
72+
Changed map[string]int `json:"changed"`
73+
Ok map[string]int `json:"ok"`
74+
Failures map[string]int `json:"failures"`
75+
Skipped map[string]int `json:"skipped"`
76+
}

‎pkg/ansible/runner/internal/inputdir/inputdir.go

Lines changed: 123 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,123 @@
1+
// Copyright 2018 The Operator-SDK Authors
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package inputdir
16+
17+
import (
18+
"encoding/json"
19+
"fmt"
20+
"io/ioutil"
21+
"os"
22+
"path/filepath"
23+
24+
"github.com/sirupsen/logrus"
25+
)
26+
27+
// InputDir represents an input directory for ansible-runner.
28+
type InputDir struct {
29+
Path string
30+
PlaybookPath string
31+
Parameters map[string]interface{}
32+
EnvVars map[string]string
33+
Settings map[string]string
34+
}
35+
36+
// makeDirs creates the required directory structure.
37+
func (i *InputDir) makeDirs() error {
38+
for _, path := range []string{"env", "project", "inventory"} {
39+
fullPath := filepath.Join(i.Path, path)
40+
err := os.MkdirAll(fullPath, os.ModePerm)
41+
if err != nil {
42+
logrus.Errorf("unable to create directory %v", fullPath)
43+
return err
44+
}
45+
}
46+
return nil
47+
}
48+
49+
// addFile adds a file to the given relative path within the input directory.
50+
func (i *InputDir) addFile(path string, content []byte) error {
51+
fullPath := filepath.Join(i.Path, path)
52+
err := ioutil.WriteFile(fullPath, content, 0644)
53+
if err != nil {
54+
logrus.Errorf("unable to write file %v", fullPath)
55+
}
56+
return err
57+
}
58+
59+
// Write commits the object's state to the filesystem at i.Path.
60+
func (i *InputDir) Write() error {
61+
paramBytes, err := json.Marshal(i.Parameters)
62+
if err != nil {
63+
return err
64+
}
65+
envVarBytes, err := json.Marshal(i.EnvVars)
66+
if err != nil {
67+
return err
68+
}
69+
settingsBytes, err := json.Marshal(i.Settings)
70+
if err != nil {
71+
return err
72+
}
73+
74+
err = i.makeDirs()
75+
if err != nil {
76+
return err
77+
}
78+
79+
err = i.addFile("env/envvars", envVarBytes)
80+
if err != nil {
81+
return err
82+
}
83+
err = i.addFile("env/extravars", paramBytes)
84+
if err != nil {
85+
return err
86+
}
87+
err = i.addFile("env/settings", settingsBytes)
88+
if err != nil {
89+
return err
90+
}
91+
92+
// If ansible-runner is running in a python virtual environment, propagate
93+
// that to ansible.
94+
venv := os.Getenv("VIRTUAL_ENV")
95+
hosts := "localhost ansible_connection=local"
96+
if venv != "" {
97+
hosts = fmt.Sprintf("%s ansible_python_interpreter=%s", hosts, filepath.Join(venv, "bin/python"))
98+
}
99+
err = i.addFile("inventory/hosts", []byte(hosts))
100+
if err != nil {
101+
return err
102+
}
103+
104+
if i.PlaybookPath != "" {
105+
f, err := os.Open(i.PlaybookPath)
106+
if err != nil {
107+
logrus.Errorf("failed to open playbook file %v", i.PlaybookPath)
108+
return err
109+
}
110+
defer f.Close()
111+
112+
playbookBytes, err := ioutil.ReadAll(f)
113+
if err != nil {
114+
return err
115+
}
116+
117+
err = i.addFile("project/playbook.yaml", playbookBytes)
118+
if err != nil {
119+
return err
120+
}
121+
}
122+
return nil
123+
}

0 commit comments

Comments
 (0)