<commit_before>// Copyright 2019 The Cloud Robotics Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

// Package main runs a multiplexing HTTP relay server.
//
// It exists to make HTTP endpoints on robots accessible without a public
// endpoint. It binds to a public endpoint accessible by both client and
// backend and works together with a relay client that's colocated with the
// backend.
//
//          lan   |    internet      |               lan
//                |                  |
//        client ---> relay server <--- relay client ---> backend
//                |                  |
//             firewall           firewall
//
// The relay server is multiplexing: It allows multiple relay clients to
// connect under unique names, each handling requests for a subpath of /client.
//
// Sequence of operations:
//   * Client makes request on /client/$foo/$request.
//   * Relay server assigns an ID and stores request (with path $request) in
//     memory. It keeps the client's request pending.
//   * Relay client requests /server/request?server=$foo
//   * Relay server responds with stored request (or timeout if no request comes
//     in within the next 30 sec).
//   * Relay client makes the stored request to backend.
//   * Backend replies.
//   * Relay client posts backend's reply to /server/response.
//   * Relay server responds to client's request with backend's reply.
//
// For some requests (eg kubectl exec), the backend responds with
// 101 Switching Protocols, resulting in the following operations.
//   * Relay server responds to client's request with backend's 101 reply.
//   * Client sends bytes from stdin to the relay server.
//   * Relay client requests /server/requeststream?id=$id.
//   * Relay server responds with stdin bytes from client.
//   * Relay client sends stdin bytes to backend.
//   * Backend sends stdout bytes to relay client.
//   * Relay client posts stdout bytes to /server/response.
//   * Relay server sends stdout bytes to the client.
//
// This simplified graphic shows the back-and-forth for an `exec` request:
//
//        client ---> relay server <--- relay client ---> backend
//          .     |        .         |       .               .
//          . -POST /exec->.         |       .               .
//          .     |        . <-GET /request- .               .
//          .     |        . ---- exec ----> .               .
//          .     |        .         |       . -POST /exec-> .
//          .     |        .         |       . <--- 101 ---- .
//          .     |        .<-POST /response-.               .
//          . <-- 101 ---- .         |       .               .
//          . -- stdin --> .         |       .               .
//          .     |        .<-POST /request- .               .
//          .     |        .        stream   .               .
//          .     |        . ---- stdin ---> .               .
//          .     |        .         |       . --- stdin --> .
//          .     |        .         |       . <-- stdout--- .
//          .     |        .<-POST /response-.               .
//          . <- stdout -- .         |       .               .
//          .     |        .         |       .               .
//
// The client side implementation is in ../http-relay-client.
package main

import (
	"encoding/hex"
	"flag"
	"fmt"
	"io/ioutil"
	"log"
	"math/rand"
	"net/http"
	"net/url"
	"strings"
	"time"

	pb "github.com/googlecloudrobotics/core/src/proto/http-relay"

	"github.com/golang/protobuf/proto"
	"github.com/prometheus/client_golang/prometheus/promhttp"
)

const (
	clientPrefix = "/client/"
)

var (
	port      = flag.Int("port", 80, "Port number to listen on")
	blockSize = flag.Int("block_size", 10*1024,
		"Size of i/o buffer in bytes")
)

func createId() string {
	u := make([]byte, 16)
	// err is documented as always nil
	rand.Read(u)
	return hex.EncodeToString(u)
}

func marshalHeader(h *http.Header) []*pb.HttpHeader {
	r := []*pb.HttpHeader{}
	for k, vs := range *h {
		for _, v := range vs {
			r = append(r, &pb.HttpHeader{Name: proto.String(k), Value: proto.String(v)})
		}
	}
	return r
}

func unmarshalHeader(w http.ResponseWriter, protoHeader []*pb.HttpHeader) {
	for _, h := range protoHeader {
		w.Header().Add(*h.Name, *h.Value)
	}
}

type server struct {
	b *broker
}

func newServer() *server {
	s := &server{b: newBroker()}
	go func() {
		for t := range time.Tick(10 * time.Second) {
			s.b.ReapInactiveRequests(t.Add(-30 * time.Second))
		}
	}()
	return s
}

// responseFilter enforces that there's at least one HttpResponse in the out
// channel and that the first response has a status code. It splits the status
// code, headers and body data apart so they can be written back to the client.
func responseFilter(in <-chan *pb.HttpResponse) ([]*pb.HttpHeader, int, <-chan []byte) {
	body := make(chan []byte, 1)
	firstMessage, more := <-in
	if !more {
		brokerResponses.WithLabelValues("client", "missing_message").Inc()
		body <- []byte("Received no response from relay client")
		close(body)
		return nil, http.StatusInternalServerError, body
	}
	if firstMessage.StatusCode == nil {
		brokerResponses.WithLabelValues("client", "missing_header").Inc()<commit_msg>Bump Relay timeout to 60 and improve error message.<commit_after>// Copyright 2019 The Cloud Robotics Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

// Package main runs a multiplexing HTTP relay server.
//
// It exists to make HTTP endpoints on robots accessible without a public
// endpoint. It binds to a public endpoint accessible by both client and
// backend and works together with a relay client that's colocated with the
// backend.
//
//          lan   |    internet      |               lan
//                |                  |
//        client ---> relay server <--- relay client ---> backend
//                |                  |
//             firewall           firewall
//
// The relay server is multiplexing: It allows multiple relay clients to
// connect under unique names, each handling requests for a subpath of /client.
//
// Sequence of operations:
//   * Client makes request on /client/$foo/$request.
//   * Relay server assigns an ID and stores request (with path $request) in
//     memory. It keeps the client's request pending.
//   * Relay client requests /server/request?server=$foo
//   * Relay server responds with stored request (or timeout if no request comes
//     in within the next 30 sec).
//   * Relay client makes the stored request to backend.
//   * Backend replies.
//   * Relay client posts backend's reply to /server/response.
//   * Relay server responds to client's request with backend's reply.
//
// For some requests (eg kubectl exec), the backend responds with
// 101 Switching Protocols, resulting in the following operations.
//   * Relay server responds to client's request with backend's 101 reply.
//   * Client sends bytes from stdin to the relay server.
//   * Relay client requests /server/requeststream?id=$id.
//   * Relay server responds with stdin bytes from client.
//   * Relay client sends stdin bytes to backend.
//   * Backend sends stdout bytes to relay client.
//   * Relay client posts stdout bytes to /server/response.
//   * Relay server sends stdout bytes to the client.
//
// This simplified graphic shows the back-and-forth for an `exec` request:
//
//        client ---> relay server <--- relay client ---> backend
//          .     |        .         |       .               .
//          . -POST /exec->.         |       .               .
//          .     |        . <-GET /request- .               .
//          .     |        . ---- exec ----> .               .
//          .     |        .         |       . -POST /exec-> .
//          .     |        .         |       . <--- 101 ---- .
//          .     |        .<-POST /response-.               .
//          . <-- 101 ---- .         |       .               .
//          . -- stdin --> .         |       .               .
//          .     |        .<-POST /request- .               .
//          .     |        .        stream   .               .
//          .     |        . ---- stdin ---> .               .
//          .     |        .         |       . --- stdin --> .
//          .     |        .         |       . <-- stdout--- .
//          .     |        .<-POST /response-.               .
//          . <- stdout -- .         |       .               .
//          .     |        .         |       .               .
//
// The client side implementation is in ../http-relay-client.
package main

import (
	"encoding/hex"
	"flag"
	"fmt"
	"io/ioutil"
	"log"
	"math/rand"
	"net/http"
	"net/url"
	"strings"
	"time"

	pb "github.com/googlecloudrobotics/core/src/proto/http-relay"

	"github.com/golang/protobuf/proto"
	"github.com/prometheus/client_golang/prometheus/promhttp"
)

const (
	clientPrefix           = "/client/"
	inactiveRequestTimeout = 60 * time.Second
)

var (
	port      = flag.Int("port", 80, "Port number to listen on")
	blockSize = flag.Int("block_size", 10*1024,
		"Size of i/o buffer in bytes")
)

func createId() string {
	u := make([]byte, 16)
	// err is documented as always nil
	rand.Read(u)
	return hex.EncodeToString(u)
}

func marshalHeader(h *http.Header) []*pb.HttpHeader {
	r := []*pb.HttpHeader{}
	for k, vs := range *h {
		for _, v := range vs {
			r = append(r, &pb.HttpHeader{Name: proto.String(k), Value: proto.String(v)})
		}
	}
	return r
}

func unmarshalHeader(w http.ResponseWriter, protoHeader []*pb.HttpHeader) {
	for _, h := range protoHeader {
		w.Header().Add(*h.Name, *h.Value)
	}
}

type server struct {
	b *broker
}

func newServer() *server {
	s := &server{b: newBroker()}
	go func() {
		for t := range time.Tick(10 * time.Second) {
			s.b.ReapInactiveRequests(t.Add(-1 * inactiveRequestTimeout))
		}
	}()
	return s
}

// responseFilter enforces that there's at least one HttpResponse in the out
// channel and that the first response has a status code. It splits the status
// code, headers and body data apart so they can be written back to the client.
func responseFilter(in <-chan *pb.HttpResponse) ([]*pb.HttpHeader, int, <-chan []byte) {
	body := make(chan []byte, 1)
	firstMessage, more := <-in
	if !more {
		brokerResponses.WithLabelValues("client", "missing_message").Inc()
		body <- []byte(fmt.Sprintf("Timeout after %v, either the backend request took too long or the relay client died", inactiveRequestTimeout))
		close(body)
		return nil, http.StatusInternalServerError, body
	}
	if firstMessage.StatusCode == nil {
		brokerResponses.WithLabelValues("client", "missing_header").Inc()
{'fraction_non_alphanumeric': 0.13048707155742634, 'fraction_numerical': 0.00661455201443175, 'mean_word_length': 1.7425206124852768, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 22, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1349799', 'n_tokens_mistral': 3492, 'n_tokens_neox': 3284, 'n_words': 1112}