[[380142]]

必要

次のような問題を通常はどうやって解決するのでしょうか?

新しいポッドを作成し、別のポッドからこの新しいポッドにファイルをコピーして分析を行いたいのですが、どうすればよいでしょうか？

`kubectl cp` を使用してファイルをコピーするのと同様に、プロジェクト内の Pod 間でファイルのコピーを実装するにはどうすればよいですか?

新しい Pod はインスタンス Pod と PVC を共有する必要がありますか? それとも、認証コンテキストを使用してコマンドラインを実行する kubectl インスタンスでラップする必要がありますか?

導入

この記事では、Kubernetes の exec websocket インターフェースを tar と組み合わせて使用​​し、ファイル ツリーをデータ ストリームに圧縮してから解凍して復元することで、2 つの Pod 間でファイルをコピーする方法を説明します。

exec インターフェースの詳細については、https://www.cnblogs.com/a00ium/p/10905279.html を参照してください。

画像をご覧ください。

プロセスの説明

• まず、コルーチン間のシグナル通知のためにシグナルチャネルが初期化されます。シグナルを受信したコルーチンは、一時停止、ループの終了、チャネルのクローズなどの操作を実行します。
• 初期化プロセスでは、バイト配列 `[]byte` であるデータチャネル `srcStdOutCh` を使用します。このチャネルは、ソース Pod の標準出力をチャネルに入力するために使用され、宛先 Pod の標準入力に送信されるデータは、このデータチャネルから読み取られます。
• exec インターフェースのアクセス アドレス (クラスター接続、トークン)、tar 圧縮コマンド、標準入出力、tty、ポッド名、コンテナ名、その他のパラメータを連結します。`tar czf - /var/log/xxx.log` は、ファイル ツリー構造がデータ ストリームに圧縮されることを示します。
• WebSocket の Dialer メソッドが呼び出され、ソース Pod コンテナとの WebSocket 接続が確立され、goroutine が開始されて標準出力がデータ チャネル srcStdOutCh に書き込まれます。
• ソース pod exec インターフェイスを参照して、宛先 pod exec アクセス接続を構築します。`tar xzf - -C /tmp` は、標準入力からデータ ストリームを読み取り、それをファイル ツリー構造に解凍することを示します (注: 解凍されたファイル ディレクトリ ツリー構造が含まれます)。
• 宛先PodとのWSS接続を確立し、データチャネルsrcStdOutChからソースPodの標準出力を読み取り、宛先Podの標準入力に書き込むgoroutineを起動します。データチャネルからの読み取りがタイムアウトした場合、データ転送が完了したことを意味します。この時点で、宛先コンテナへのデータ入力を停止し、メインgoroutineに終了通知を送信して、ソースPodとのWSS接続を閉じます。

予防

• wesocketがソースPodに接続すると、標準出力に空のデータや、tarコマンドの出力など、他の干渉データが含まれる場合があります。そのため、データフィルタリングのためにデータを受信する際には、フィルターコールバック関数を渡す必要があります。
• 宛先コンテナにデータを送信する際、送信元コンテナが受信した最初のバイトを削除する必要があります。このバイトは通常1で、標準出力フラグを表します。宛先コンテナにデータを送信する際には不要です。
• データを送信するときは、標準入力に送信されていることを示すために、最初のバイトを 0 に設定する必要があります。

参照コード

cp.go

 /*
概要: 1. バッファリングされていないチャネルでは、書き込み前に読み取りが必要です。2. WebSocket の ReadMessage メソッドはブロックされています。読み取りを中断するには、接続を閉じてエラーをキャッチします。
 /
パッケージ cpFilePod2Pod 
 
輸入 （
 「暗号/TLS」  
 「エラー」  
 「fmt」  
 "ログ"  
 「ネット/URL」  
 「正規表現 
 「文字列」  
 「同期」  
 "時間"   
 
 「github.com/gorilla/websocket」  
 ） 
 
 // フィルタコールバック関数を定義する
type filterCallback func(入力文字列) bool 
 
 // ミューテックスロック付きのWebSocket接続オブジェクト
WsConn構造体型{
 Conn *websocket.Conn
 mu sync.Mutex
 } 
 
 // 文字列を送信し、自動的に改行文字を追加します
func (self *WsConn) Send(送信者文字列, str文字列) {
 self.mu.Lock()
 self.mu.Unlock() を延期する
Kubernetes exec websocket インターフェースを使用してデータを送信する場合、データが標準入力に送信されていることを示すために、最初のバイトを 0 に設定する必要があります。
データ := []バイト{0}
データ = append(data, []byte(str+ "\n" )...)
 err := self.Conn.WriteMessage(websocket.BinaryMessage, data) // バイナリデータ型を送信
err != nil の場合 {
 log.Printf( "送信エラー、%s" , err.Error())
 }
 log.Printf( "%s, データ:%s, バイ​​ト数:%+v" , 送信者, str, []byte(str+ "\n" ))
 } 
 
 // 改行文字を追加せずに文字列を送信します。内部のバイト フィルタリングなどの操作が実行されます。
 func (self *WsConn) SendWithFilter(送信者文字列, str文字列) {
 self.mu.Lock()
 self.mu.Unlock() を延期する
// log.Printf( "宛先コンテナにデータを送信しました:%s" , str)
 str = strings.ReplaceAll(str, "\r\n" , "\n" ) // /r=13, /n=10、Windows の改行文字を Linux の改行文字に変換します
//最初のバイト (標準出力 1、バイト: [0 1 ...]) を削除します。ソース コンテナからのバイト出力の最初のビットは標準出力 1 を識別するため、このフラグは宛先コンテナにバイトを送信するときに削除する必要があります。
 //WebSocket 接続が確立されると、データを送信するときはバイト バッファーの最初のバイトを stdin (buf[0] = 0) に設定する必要があり、データを受信するときは最初のバイトをチェックして、stdout (buf[0] = 1) または stderr (buf[0] = 2) のいずれかを使用する必要があります。
 strByte := append([]byte(str)[:0], []byte(str)[1:]...)
データ := []バイト{0}
データ = append(データ、strByte...)
エラー:= self.Conn.WriteMessage(websocket.BinaryMessage, データ)
 log.Printf( "宛先コンテナの標準入力にデータを送信しました:\n%s、バイト数:%d、バイト数:%+v" 、文字列(データ)、長さ(データ)、データ)
 err != nil の場合 {
 log.Printf( "送信エラー、%s" , err.Error())
 }
 } 
 
 // 接続からデータ ストリームを取得し、バイト配列チャネルに書き込み、内部でフィルター (コールバック関数) を実行します。
 func (self *WsConn) Receive(receiver string, ch chan []byte, filter filterCallback) error {
 self.mu.Lock()
 self.mu.Unlock() を延期する
msgType, msgByte, err := self.Conn.ReadMessage() // ブロッキング読み取り、タイプ 2 はバイナリデータ、1 はテキスト、-1 は接続が閉じられていることを示します: websocket: close 1000 (正常)
 log.Printf( "%s, 読み取ったデータ:%s, タイプ:%d, バイト数:%d, バイト数:%+v" , レシーバー, string(msgByte), msgType, len(msgByte), msgByte)
 err != nil の場合 {
 log.Printf( "%s, 読み取りエラー、%s" , レシーバー, err.Error())
エラーを返す
}
フィルター(文字列(msgByte)) && len(msgByte) > 1 の場合 {
 ch <- メッセージバイト
}それ以外{
 log.Printf( "%s, データが満たされていないため、データを直接破棄します。文字:%s、バイト数:%d、バイト数:%v" , レシーバー, string(msgByte), len(msgByte), msgByte)
 }
 nilを返す
} 
 
 func NewWsConn(ホスト文字列、パス文字列、パラメータマップ[文字列]文字列、ヘッダーマップ[文字列][]文字列) (*websocket.Conn, error) {
パラメータ配列:=[]文字列{}
 k , v := 範囲パラメータ {
 paramArray = append(paramArray, fmt.Sprintf( "%s=%s" , k, v))
 }
 u := url.URL{スキーム: "wss" , ホスト: host, パス: path, RawQuery: strings. Join (paramArray, "&" )}
 log.Printf( "API:%s" , u.String())
ダイヤラー:= websocket.Dialer{TLSClientConfig: &tls.Config{RootCAs: nil, InsecureSkipVerify: true }}
 conn, _, err := dialer.Dial(u.String(), ヘッダー)
 err != nil の場合 {
 nilを返します。errors.New(fmt.Sprintf( "接続エラー:%s" , err.Error()))
 }
 conn、nilを返す
} 
 
 //コア: tar -cf - フォルダー構造を持つデータをデータ ストリームに変換し、tar -xf - データ ストリームを Linux ファイル システムに変換します。
 CpPod2Pod()関数 {
 // メイン関数に終了できることを通知するシグナルチャネル
signalExit := make(chan bool, 1)
遅延クローズ（signalExit） 
 
 //ターゲットコンテナにデータを送信しないように信号を送信します
signalStopDstSend := make(chan bool, 1)
遅延クローズ(signalStopDstSend) 
 
 // ソースコンテナからのデータの読み取りを防止するためのシグナルを送信します
signalStopSrcRead := make(chan bool, 1)
遅延クローズ(signalStopSrcRead) 
 
 //宛先コンテナからデータを読み取らないように信号を送信します
signalStopDstRead := make(chan bool, 1)
遅延クローズ(signalStopDstRead) 
 
 // 印刷しない出力データを宛先コンテナーに送信します。
 signalStopPrintDstStdout := make(chan bool, 1)
遅延クローズ(signalStopPrintDstStdout) 
 
 //ポッドに接続
ホスト := "172.16.xxx.xxx:6443"  
トークン := "xxx"  
 headers := map[string][]string{ "authorization" : {fmt.Sprintf( "Bearer %s" , token)}} 
 
パスソース:= "/api/v1/namespaces/xxx/pods/xxx/exec"  
 commandSrc := "tar&command=czf&command=-&command=/var/log/mysql/slow.log" //tar czf - ソースファイル
paraSrc := map[string]string{ "stdout" : "1" , "stdin" : "0" , "stderr" : "1" , "tty" : "0" , "container" : "xxx" , "command" : commandSrc}
 srcConn, err := NewWsConn(ホスト, pathSrc, paraSrc, ヘッダー)
 err != nil の場合 {
 log.Printf( "ソースポッド接続エラー、%s" 、 err.Error())
 } 
 
パス宛先:= "/api/v1/namespaces/xxx/pods/xxx/exec"  
 commandDst := "tar&command=xzf&command=-&command=-C&command=/tmp" // tar xzf - -C /tmp
 // paraDst := map[string]string{ "stdout" : "1" , "stdin" : "1" , "stderr" : "1" , "tty" : "0" , "container" : "xxx" , "command" : commandDst}
 paraDst := map[string]string{ "stdout" : "0" , "stdin" : "1" , "stderr" : "0" , "tty" : "0" , "container" : "xxx" , "command" : commandDst} //宛先Podの標準出力とエラー出力をオフにする
dstConn, err := NewWsConn(ホスト, pathDst, paraDst, ヘッダー)
 err != nil の場合 {
 log.Printf( "宛先ポッド接続エラー、%s" 、 err.Error())
 } 
 
 wsSrc := WsConn{
コネチカット州: srcConn、
 } 
 
 wsDst := WsConn{
コネチカット州: dstConn、
 } 
 
 srcConn.Close ()を延期する
dstConn.Close ()を延期する
 
 srcStdOutCh := make(chan []byte, 2048)
 dstStdOutCh := make(chan []byte)
遅延クローズ(srcStdOutCh)
遅延クローズ(dstStdOutCh) 
 
 // ソースコンテナの標準出力をデータチャネルに受信する
ゴー関数() {
私 := 1
のために{
 log.Printf( "%d回目、ソースコンテナから標準出力を読み取りました" , i)
 i++
 // ソース コンテナーの標準出力から不要なデータをフィルター処理するための匿名フィルター コールバック メソッドを定義します。
 err := wsSrc.Receive( "ソースコンテナ" , srcStdOutCh, func(入力文字列) bool {
入力 == "cat /var/log/mysql/slow.log"の場合{
戻る 間違い 
 // }それ以外の場合、一致する場合、_ := regexp.MatchString( "root@(.+)#" , input); 一致 {
 //戻る 間違い 
 // }それ以外の場合、一致する場合、_ := regexp.MatchString( "cat /(.+).log" , input); 一致 {
 //戻る 間違い 
 // }それ以外の場合、一致する場合、_ := regexp.MatchString( "cat /tmp/(.+)" , input); 一致 {
 //戻る 間違い 
 }それ以外の場合、一致する場合、_ := regexp.MatchString( "tar: 先頭の(.+)を削除します" , input); 一致 {
戻る 間違い 
 } else if len(input) == 0 { // 空のメッセージをフィルタリングする
// log.Printf( "標準エラー出力が読み取られました" )
戻る 間違い 
 }
戻る 真実 
 })
 err != nil の場合 {
 log.Printf( "ソースコンテナからの標準出力の読み取りに失敗しました" )
 // シグナル終了 <- true  
壊す
}
 // time.Sleep ( time .マイクロ秒 * 100)
 }
 }() 
 
 /* 宛先コンテナの標準出力を読み取るために、ここでは並行コルーチンを有効にできないことに注意してください。これを有効にすると、データ送信中のコルーチンとロックの競合が発生し、宛先コンテナへのデータ送信がブロックされる可能性があります。 */
 // 宛先コンテナから標準出力を取得し、データチャネルに格納します
// ゴー関数() {
 // i := 0
 //のために{
 // このフィルターは直接trueを返します。これは単なるプレースホルダーです。
 // err := wsDst.Receive( "宛先コンテナ" , dstStdOutCh, func(入力文字列) bool {
 //戻る 真実 
 // })
 // err != nil の場合 {
 // log.Printf( "宛先コンテナからのデータの読み取りに失敗しました" )
 // 壊す
// }
 // wsDst.Send()
 // time.Sleep ( time .マイクロ秒 * 100000)
 // }
 // log.Printf( "宛先コンテナからデータを読み取り中、%d 回目の反復" , i)
 // i++
 // }() 
 
 // データ チャネルから宛先コンテナの標準出力を読み取り、印刷します。
 // ゴー関数() {
 // BreakPrintDstPodStdout:
 //のために{
 //選択{
 //ケースデータ := <-dstStdOutCh:
 // log.Printf( "宛先コンテナの標準出力:%s" , string(data))
 // // time.Sleep ( time .マイクロ秒 * 200)
 //ケース<-signalStopPrintDstStdout:
 // log.Printf( "シグナルを受信しました。宛先コンテナの標準出力への出力を停止します" )
 // //閉じる(dataOutput)
 //閉じる(dataCh)
 // // signalStopRead <- true  
 // log.Printf( "読み取り停止信号を送信しました" )
 // //閉じる(dataOutput)
 //閉じる(dataCh)
 // BreakPrintDstPodStdout を中断する
// }
 // // time.Sleep ( time .マイクロ秒 * 100)
 // }
 // }() 
 
 // ソース コンテナの標準出力データ チャネルからデータを取得し、それを宛先コンテナの標準入力に送信します。
 // タイムアウト期間を定義する
タイムアウト秒数:= 3
タイマー := time .NewTimer( time . Second * time . Duration(timeOutSecond))
ブレイク2メイン:
のために{
選択{
ケースデータ := <-srcStdOutCh:
 wsDst.SendWithFilter( "宛先コンテナに送信する" , string(data))
 // time.Sleep ( time .ミリ秒 * 200)
 timer.Reset(時間.秒*時間. Duration(timeOutSecond))
ケース<-timer.C:
 // time .Sleep( time . Second * 5)
 log.Printf( "================== ソースコンテナの標準出力、新しいデータなし、タイムアウト、宛先コンテナへのデータ送信を停止 ==================" )
 // log.Printf( "シグナル送信: 宛先コンテナの標準出力への出力を停止します" )
 // signalStopPrintDstStdout <- true  
 log.Printf( "シグナル送信: ソースコンテナからのデータの読み取りを停止します" )
 wsSrc.Conn.Close ()
 // log.Printf( "シグナル送信: 宛先コンテナからのデータの読み取りを停止します" )
 // wsDst.Conn.Close ()
 log.Printf( "シグナルが送信されました: メイン関数は終了できます。" )
シグナル終了 <- true  
 log.Printf( "すべての信号が送信されました" )
 log.Printf( "================== ループ終了 ===================" )
ブレークBreak2Main
 }
 // time.Sleep ( time .マイクロ秒 * 1000)
 } 
 
 // signalStopRead <- true  
 <-signalExit // 信号が受信されるまでチャネルをブロックします。
 // signalStopRead <- true  
 log.Printf( "メイン関数はシグナルを受信し、終了の準備をしています" )。
 //閉じる(dataCh)
 // time .Sleep(時間.秒)
 //閉じる（データ出力）
 // time .Sleep(時間.秒)
 // {}を選択
}

cp_test.go

パッケージ cpFilePod2Pod 
 
輸入 （
 "ログ"  
 「テスト」  
 ） 
 
 // Go test -race -test.run TestCpPod2Pod このディレクトリに移動してテストを実行します。
 TestCpPod2Pod関数(t *テスト.T) {
 log.Printf( "テストを開始しています" )
 CpPod2Pod()
 }

参考結果:
ソースコンテナ:
 root@xxx-mysql-0:/var/log/mysql# md5sum slow.log
 16577613b6ea957ecb5d9d5e976d9c50 遅い.log
宛先コンテナ:
 root@xxx-75bdcdb8cf-hq9wf:/tmp/var/log/mysql# md5sum slow.log
 16577613b6ea957ecb5d9d5e976d9c50 遅い.log

参考資料

Kubernetes exec API 接続分析: https://www.cnblogs.com/a00ium/p/10905279.html

kubernetes-client-go-implementation-kubectl-copy:https://ica10888.com/2019/08/31/kubernetes-client-go-%E5%AE%9E%E7%8E%B0-kubectl-copy.html