+func (gw *Gateway) maintainTunnel(addr string) {
+ for ; ; time.Sleep(5 * time.Second) {
+ err := gw.runTunnel(addr)
+ gw.Log.Printf("runTunnel: %s", err)
+ }
+}
+
+// runTunnel connects to controller and sets up a tunnel through
+// which controller can connect to the gateway server at the given
+// addr.
+func (gw *Gateway) runTunnel(addr string) error {
+ ctx := auth.NewContext(context.Background(), auth.NewCredentials(gw.ArvadosClient.AuthToken))
+ arpc := rpc.NewConn("", &url.URL{Scheme: "https", Host: gw.ArvadosClient.APIHost}, gw.ArvadosClient.Insecure, rpc.PassthroughTokenProvider)
+ tun, err := arpc.ContainerGatewayTunnel(ctx, arvados.ContainerGatewayTunnelOptions{
+ UUID: gw.ContainerUUID,
+ AuthSecret: gw.AuthSecret,
+ })
+ if err != nil {
+ return fmt.Errorf("error creating gateway tunnel: %s", err)
+ }
+ mux, err := yamux.Client(tun.Conn, nil)
+ if err != nil {
+ return fmt.Errorf("error setting up mux client end: %s", err)
+ }
+ if url := tun.Header.Get("X-Arvados-Internal-Url"); url != "" && gw.UpdateTunnelURL != nil {
+ gw.UpdateTunnelURL(url)
+ }
+ for {
+ muxconn, err := mux.AcceptStream()
+ if err != nil {
+ return err
+ }
+ gw.Log.Printf("tunnel connection %d started", muxconn.StreamID())
+ go func() {
+ defer muxconn.Close()
+ gwconn, err := net.Dial("tcp", addr)
+ if err != nil {
+ gw.Log.Printf("tunnel connection %d: error connecting to %s: %s", muxconn.StreamID(), addr, err)
+ return
+ }
+ defer gwconn.Close()
+ var wg sync.WaitGroup
+ wg.Add(2)
+ go func() {
+ defer wg.Done()
+ _, err := io.Copy(gwconn, muxconn)
+ if err != nil {
+ gw.Log.Printf("tunnel connection %d: mux end: %s", muxconn.StreamID(), err)
+ }
+ gwconn.Close()
+ }()
+ go func() {
+ defer wg.Done()
+ _, err := io.Copy(muxconn, gwconn)
+ if err != nil {
+ gw.Log.Printf("tunnel connection %d: gateway end: %s", muxconn.StreamID(), err)
+ }
+ muxconn.Close()
+ }()
+ wg.Wait()
+ gw.Log.Printf("tunnel connection %d finished", muxconn.StreamID())
+ }()
+ }
+}
+