Skip to content

Commit

Permalink
Add streaming command support.
Browse files Browse the repository at this point in the history
Add options
- `stream-stdout-in-response`
- `stream-stdout-in-response-on-error`
- `stream-command-kill-grace-period-seconds`

to allow defining webhooks which dynamically stream large content back to the
requestor. This allows the creation of download endpoints from scripts, i.e.
running a `git archive` command or a database dump from a docker container,
without needing to buffer up the original.
  • Loading branch information
wrouesnel committed Mar 1, 2019
1 parent 0aa7395 commit 08fc28b
Show file tree
Hide file tree
Showing 8 changed files with 547 additions and 111 deletions.
3 changes: 3 additions & 0 deletions docs/Hook-Definition.md
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,9 @@ Hooks are defined as JSON objects. Please note that in order to be considered va
* `response-headers` - specifies the list of headers in format `{"name": "X-Example-Header", "value": "it works"}` that will be returned in HTTP response for the hook
* `include-command-output-in-response` - boolean whether webhook should wait for the command to finish and return the raw output as a response to the hook initiator. If the command fails to execute or encounters any errors while executing the response will result in 500 Internal Server Error HTTP status code, otherwise the 200 OK status code will be returned.
* `include-command-output-in-response-on-error` - boolean whether webhook should include command stdout & stderror as a response in failed executions. It only works if `include-command-output-in-response` is set to `true`.
* `stream-stdout-in-response` - boolean (exclusive with `include-command-output-in-response` and `include-command-output-in-response-on-error`) that will stream the output of a command in the response if the command writes any data to standard output before exiting non-zero.
* `stream-stderr-in-response-on-error` - boolean whether the webhook should send the stream of stderr on error. Only effective if `stream-stdout-in-response` is being used.
* `stream-command-kill-grace-period-seconds` - float number of seconds to wait after trying to kill a stream command with SIGTERM before sending SIGKILL. Default is 0 (do not wait).
* `parse-parameters-as-json` - specifies the list of arguments that contain JSON strings. These parameters will be decoded by webhook and you can access them like regular objects in rules and `pass-arguments-to-command`.
* `pass-arguments-to-command` - specifies the list of arguments that will be passed to the command. Check [Referencing request values page](Referencing-Request-Values.md) to see how to reference the values from the request. If you want to pass a static string value to your command you can specify it as
`{ "source": "string", "name": "argumentvalue" }`
Expand Down
3 changes: 3 additions & 0 deletions hook/hook.go
Original file line number Diff line number Diff line change
Expand Up @@ -420,6 +420,9 @@ type Hook struct {
ResponseHeaders ResponseHeaders `json:"response-headers,omitempty"`
CaptureCommandOutput bool `json:"include-command-output-in-response,omitempty"`
CaptureCommandOutputOnError bool `json:"include-command-output-in-response-on-error,omitempty"`
StreamCommandStdout bool `json:"stream-stdout-in-response,omitempty"`
StreamCommandStderrOnError bool `json:"stream-stderr-in-response-on-error,omitempty"`
StreamCommandKillGraceSecs float64 `json:"stream-command-kill-grace-period-seconds,omitempty"`
PassEnvironmentToCommand []Argument `json:"pass-environment-to-command,omitempty"`
PassArgumentsToCommand []Argument `json:"pass-arguments-to-command,omitempty"`
PassFileToCommand []Argument `json:"pass-file-to-command,omitempty"`
Expand Down
61 changes: 49 additions & 12 deletions test/hookecho.go
Original file line number Diff line number Diff line change
Expand Up @@ -5,13 +5,58 @@ package main
import (
"fmt"
"os"
"strconv"
"strings"
"strconv"
"io"
)

func checkPrefix(prefixMap map[string]struct{}, prefix string, arg string) bool {
if _, found := prefixMap[prefix]; found {
fmt.Printf("prefix specified more then once: %s", arg)
os.Exit(-1)
}

if strings.HasPrefix(arg, prefix) {
prefixMap[prefix] = struct{}{}
return true
}

return false
}

func main() {
var outputStream io.Writer
outputStream = os.Stdout
seenPrefixes := make(map[string]struct{})
exit_code := 0

for _, arg := range os.Args[1:] {
if checkPrefix(seenPrefixes, "stream=", arg) {
switch arg {
case "stream=stdout":
outputStream = os.Stdout
case "stream=stderr":
outputStream = os.Stderr
case "stream=both":
outputStream = io.MultiWriter(os.Stdout, os.Stderr)
default:
fmt.Printf("unrecognized stream specification: %s", arg)
os.Exit(-1)
}
} else if checkPrefix(seenPrefixes, "exit=", arg) {
exit_code_str := arg[5:]
var err error
exit_code_conv, err := strconv.Atoi(exit_code_str)
exit_code = exit_code_conv
if err != nil {
fmt.Printf("Exit code %s not an int!", exit_code_str)
os.Exit(-1)
}
}
}

if len(os.Args) > 1 {
fmt.Printf("arg: %s\n", strings.Join(os.Args[1:], " "))
fmt.Fprintf(outputStream, "arg: %s\n", strings.Join(os.Args[1:], " "))
}

var env []string
Expand All @@ -22,16 +67,8 @@ func main() {
}

if len(env) > 0 {
fmt.Printf("env: %s\n", strings.Join(env, " "))
fmt.Fprintf(outputStream, "env: %s\n", strings.Join(env, " "))
}

if (len(os.Args) > 1) && (strings.HasPrefix(os.Args[1], "exit=")) {
exit_code_str := os.Args[1][5:]
exit_code, err := strconv.Atoi(exit_code_str)
if err != nil {
fmt.Printf("Exit code %s not an int!", exit_code_str)
os.Exit(-1)
}
os.Exit(exit_code)
}
os.Exit(exit_code)
}
31 changes: 31 additions & 0 deletions test/hooks.json.tmpl
Original file line number Diff line number Diff line change
Expand Up @@ -204,5 +204,36 @@
"name": "passed"
}
],
},
{
"id": "stream-stdout-in-response",
"pass-arguments-to-command": [
{
"source": "string",
"name": "exit=0"
},
{
"source": "string",
"name": "stream=both"
}
],
"execute-command": "{{ .Hookecho }}",
"stream-stdout-in-response": true
},
{
"id": "stream-stderr-in-response-on-error",
"pass-arguments-to-command": [
{
"source": "string",
"name": "exit=1"
},
{
"source": "string",
"name": "stream=stderr"
}
],
"execute-command": "{{ .Hookecho }}",
"stream-stdout-in-response": true,
"stream-stderr-in-response-on-error": true
}
]
21 changes: 20 additions & 1 deletion test/hooks.yaml.tmpl
Original file line number Diff line number Diff line change
Expand Up @@ -113,4 +113,23 @@

- id: warn-on-space
execute-command: '{{ .Hookecho }} foo'
include-command-output-in-response: true
include-command-output-in-response: true

- id: stream-stdout-in-response
execute-command: '{{ .Hookecho }}'
stream-stdout-in-response: true
pass-arguments-to-command:
- source: string
name: exit=0
- source: string
name: stream=both

- id: stream-stderr-in-response-on-error
execute-command: '{{ .Hookecho }}'
stream-stdout-in-response: true
stream-stderr-in-response-on-error: true
pass-arguments-to-command:
- source: string
name: exit=1
- source: string
name: stream=stderr
107 changes: 107 additions & 0 deletions test/hookstream/hookstream.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,107 @@
// Hook Stream is a simple utility for testing Webhook streaming capability. It spawns a TCP server on execution
// which echos all connections to its stdout until it receives the string EOF.

package main

import (
"fmt"
"os"
"strings"
"strconv"
"io"
"net"
"bufio"
)

func checkPrefix(prefixMap map[string]struct{}, prefix string, arg string) bool {
if _, found := prefixMap[prefix]; found {
fmt.Printf("prefix specified more then once: %s", arg)
os.Exit(-1)
}

if strings.HasPrefix(arg, prefix) {
prefixMap[prefix] = struct{}{}
return true
}

return false
}

func main() {
var outputStream io.Writer
outputStream = os.Stdout
seenPrefixes := make(map[string]struct{})
exit_code := 0

for _, arg := range os.Args[1:] {
if checkPrefix(seenPrefixes, "stream=", arg) {
switch arg {
case "stream=stdout":
outputStream = os.Stdout
case "stream=stderr":
outputStream = os.Stderr
case "stream=both":
outputStream = io.MultiWriter(os.Stdout, os.Stderr)
default:
fmt.Printf("unrecognized stream specification: %s", arg)
os.Exit(-1)
}
} else if checkPrefix(seenPrefixes, "exit=", arg) {
exit_code_str := arg[5:]
var err error
exit_code_conv, err := strconv.Atoi(exit_code_str)
exit_code = exit_code_conv
if err != nil {
fmt.Printf("Exit code %s not an int!", exit_code_str)
os.Exit(-1)
}
}
}

l, err := net.Listen("tcp", "localhost:0")
if err != nil {
fmt.Printf("Error starting tcp server: %v\n", err)
os.Exit(-1)
}
defer l.Close()

// Emit the address of the server
fmt.Printf("%v\n",l.Addr())

manageCh := make(chan struct{})

go func() {
for {
conn, err := l.Accept()
if err != nil {
fmt.Printf("Error accepting connection: %v\n", err)
os.Exit(-1)
}
go handleRequest(manageCh, outputStream, conn)
}
}()

<- manageCh
l.Close()

os.Exit(exit_code)
}

// Handles incoming requests.
func handleRequest(manageCh chan<- struct{}, w io.Writer, conn net.Conn) {
defer conn.Close()
bio := bufio.NewScanner(conn)
for bio.Scan() {
if line := strings.TrimSuffix(bio.Text(), "\n"); line == "EOF" {
// Request program close
select {
case manageCh <- struct{}{}:
// Request sent.
default:
// Already closing
}
break
}
fmt.Fprintf(w, "%s\n", bio.Text())
}
}

0 comments on commit 08fc28b

Please sign in to comment.